Design Patterns @ Python – Observer 觀察者模式筆記

ChrisWei · 2023年11月15日03:15

設計模式–觀察者模式(Observer Pattern)

這是 for Udemy Design Patterns in Python 課程 (講師是 Mr. Dmitri Nesteruk) 第 20 節主題為 Observer 觀察者模式的筆記，由於這位 Dmitri 老師的範例對我而言，有時會太難看不懂，因此我還有收看對岸圖靈星球論壇於油管發表的設計模式系列的 Youtube 影片來輔助學習。

Observer 觀察者模式 Image Create by Microsoft Image Creater1024×1024 196 KB

什麼是 Observer Pattern?

Observer Pattern 是一種 行為型設計模式 ,它定義物件之間的一種一對多的依賴關係,當一個物件的狀態發生改變時,所有依賴於它的物件都將得到通知並被更新。
Observer Pattern 重點速寫:
- 觀察者模式定義了一種一對多的依賴關係，使得每當一個物件狀態發生改變時，其相關的依賴物件皆得到通知並被自動更新。
  - 這樣的關係可以是當一個物件狀態是系統所關注的，因此需要通知到所有有使用到它或參考它的其他物件。
  - 也可以是當有其他(可能是一個或多個)物件是參考某個物件的值來變更其值或行為時，這樣的連動行為。
- 觀察者模式的核心是 主題（Subject） 和 觀察者（Observer） 兩個角色。
  - 主題 負責管理觀察者列表，並在其狀態發生改變時通知所有觀察者。
  - 觀察者 需要實現一個更新方法，當收到主題的通知時，調用該方法進行更新操作。
- 觀察者模式可以想像成一個事件的物件，訂閱者與事件之間的關係，訂閱者訂閱事件，事件通知訂閱者。

Observer Pattern Class Diagram

classDiagram
Direction: right
Subject <|-- ConcreteSubject
Observer <|-- ConcreteObserver1
Observer <|-- ConcreteObserver2

Subject : -observers
Subject : +attach()
Subject : +detach()
Subject : +notify()

Observer : +update()

ConcreteSubject : +getState()
ConcreteSubject : +setState()
  
ConcreteObserver1: -state
ConcreteObserver2: -state

Subject --> Observer
ConcreteSubject --> ConcreteObserver1
ConcreteSubject --> ConcreteObserver2

Class Diagram 說明:
- Subject – (事件訂閱的)對象模板(介面)
- ConcreteSubject – 具體的(事件訂閱)對象(實例 Instance)
- Observer – 觀察者(事件)模板(介面)(要觀察的屬性狀態的樣板)
- ConcreteObserver – 具體要觀察的(事件)屬性狀態的實例

Example-01

class Subject:
    def __init__(self):
        self.observers = []

    def add_observer(self, observer):
        self.observers.append(observer)

    def remove_observer(self, observer):
        self.observers.remove(observer)

    def notify_observers(self):
        for observer in self.observers:
            observer.update(self)


class Observer:
    def update(self, subject):
        pass


class ConcreteObserver1(Observer):
    def update(self, subject):
        print("ConcreteObserver1 收到通知")


class ConcreteObserver2(Observer):
    def update(self, subject):
        print("ConcreteObserver2 收到通知")


if __name__ == "__main__":
    subject = Subject()
    observer1 = ConcreteObserver1()
    observer2 = ConcreteObserver2()

    subject.add_observer(observer1)
    subject.add_observer(observer2)

    subject.notify_observers()

# 執行結果:
ConcreteObserver1 收到通知
ConcreteObserver2 收到通知

Example-02

這個範例其實是 Dmitri Nesteruk 老師講解觀察者模式的第一個範例，大致的場景如下:
1. 我們有一個主題是監理所，監理所需要知道每個人的年紀來核發駕照，以年紀來判斷可否開車
2. 我們有一個人，這個人的年紀會隨著時間改變，所以我們需要一個觀察者來觀察這個人的年紀是否改變
3. 當這個人的年紀改變時，觀察者會通知監理所，監理所會根據年紀來判斷是否可以開車
4. 當這個人的年紀未到時，監理所會觀察並發出通知還不能開車，但一旦年紀到了，監理所就不需再觀察這個人的年紀了

class Event(list):
  ''' 
  事件的集合，而事件是一個函式，所以這個集合就是一個函式的集合
  因此裡面只有一個方法，就是當 implement __call__ 來呼叫這個事件函式集合時，就會遞迴執行裡面的函式
  '''
  def __call__(self, *args, **kwargs):
    for item in self:
      # Event 傳入的參數，可以遞迴再傳入到 Event 裡面的函式
      item(*args, **kwargs)


class PropertyObservable:
  ''' 
  這是一個觀察者的基底類別，裡面有一個屬性是 property_changed，這個屬性是一個事件集合
  '''
  def __init__(self):
    self.property_changed = Event()


class Person(PropertyObservable):
  ''' 
  Person 類別，可被觀察的對象，繼承自 PropertyObservable，並且有一個屬性是 age
  '''
  def __init__(self, age=0):
    # 當 Person 類別被初始化時，會先呼叫 PropertyObservable 的 __init__ 來初始化一個事件集合
    super().__init__()
    self._age = age

  @property
  def age(self):
    return self._age

  @age.setter
  def age(self, value):
    if self._age == value:
      return
    self._age = value
    # 當 age 屬性被設定時，會呼叫 property_changed 事件集合，並傳入 age 屬性的名稱與值
    self.property_changed('age', value)


class TrafficAuthority:
  ''' 
  TrafficAuthority 類別(監理所)，初始化時會傳入一個 Person 類別，並且註冊 Person 類別的 property_changed 事件
  '''
  def __init__(self, person):
    self.person = person
    person.property_changed.append(self.person_changed)

  def person_changed(self, name, value):
    # 觀察 Person Instance 的 Age 屬性，當 Age 屬性改變時，會觸發這個函式 --> 發出通知
    if name == 'age':
      if value < 16:
        print('Sorry, you still cannot drive')
      else:
        print('Okay, you can drive now')
        # 當 Age 一過觀察臨界值時，就移除事件通知
        self.person.property_changed.remove(
          self.person_changed
        )


if __name__ == '__main__':
  p = Person()
  ta = TrafficAuthority(p)
  for age in range(14, 20):
    print(f'Setting age to {age}')
    p.age = age

# 執行結果:
Setting age to 14
Sorry, you still cannot drive
Setting age to 15
Sorry, you still cannot drive
Setting age to 16
Okay, you can drive now
Setting age to 17
Setting age to 18
Setting age to 19

Example-03

這個範例是 Dmitri Nesteruk 老師講解觀察者模式的另一個有趣範例，大致的場景如下:
1. 每個公民年滿 18 歲就有選舉權，可以參與投票，和剛剛能否開車的範例很像，這次則是能否投票
2. 同樣的我們有一個人，這個人的年紀會隨著時間改變，所以我們需要一個觀察者(協助里辦)來觀察這個人的年紀是否改變 → 年滿十八歲
3. 當這個人的年紀改變時，觀察者會通知里辦，里辦會根據這個人的年紀來判斷是否可以投票

class Event(list):
  ''' 
  事件的集合，而事件是一個函式，所以這個集合就是一個函式的集合
  因此裡面只有一個方法，就是當 implement __call__ 來呼叫這個事件函式集合時，就會遞迴執行裡面的函式
  '''  
  def __call__(self, *args, **kwargs):
    for item in self:
      item(*args, **kwargs)


class PropertyObservable:
  ''' 
  這是一個觀察者的基底類別，裡面有一個屬性是 property_changed，這個屬性是一個事件集合
  '''  
  def __init__(self):
    self.property_changed = Event()


class Person(PropertyObservable):
  ''' 
  Person 類別，可被觀察的對象，繼承自 PropertyObservable，並且有一個屬性是 age
  '''  
  def __init__(self, age=0):
    super().__init__()
    self._age = age

  @property
  def can_vote(self):
    # 能否投票是一個布林值，因此只需要判斷年紀是否大於等於 18 歲即可
    # 這個能否投票的布林屬性，就是典型的一個屬性隨另一個屬性改變而改變的情況
    return self._age >= 18

  @property
  def age(self):
    return self._age

  ''' 
  由於能否投票是一個轉變狀態，一旦過 18 歲其實就不須再繼續監控，
  而且實況是我們只需要在滿18歲可投票當下通知即可，
  不需要像剛剛範例那樣，未滿18歲前，每次都通知，
  因此我們這次不用前一個例子的作法來寫 age.setter 
  ''' 
  # @age.setter
  # def age(self, value):
  #   if self._age == value:
  #     return
  #   self._age = value
  #   self.property_changed('age', value)

  @age.setter
  def age(self, value):
    if self._age == value:
      return
    
    # 改用 old_can_vote 來記錄舊的能否投票的狀態(在改動年紀前，先記錄下來)
    old_can_vote = self.can_vote

    self._age = value
    self.property_changed('age', value)

    # 因此只有當從不能投票轉成能投票的年紀，才會進行通知
    if old_can_vote != self.can_vote:
      self.property_changed('can_vote', self.can_vote)


if __name__ == '__main__':
  def person_changed(name, value):
    if name == 'can_vote':
      print(f'Voting status changed to {value}')

  p = Person()
  p.property_changed.append(
    person_changed
  )

  for age in range(16, 21):
    print(f'Changing age to {age}')
    p.age = age

# 執行結果:
Changing age to 16
Changing age to 17
Changing age to 18
Voting status changed to True
Changing age to 19
Changing age to 20

Python callback 用法重點以及與觀察者模式比較

Python callback 是一種函式，它會 在另一個函式發生某些事件時被呼叫。callback 函式可以用來 處理事件、回傳資料或執行其他任務。
callback 的用法重點如下：
- callback 函式 必須 以參數的形式 傳遞給另一個函式。
- callback 函式 通常會在另一個函式完成某些工作後被呼叫。
- callback 函式 可以用來處理事件、回傳資料或執行其他任務。
以下是一個使用 callback 函式的範例：

def my_function(callback):
    print("我正在執行 my_function() 函式。")
    callback()

def my_callback_function():
    print("我正在執行 my_callback_function() 函式。")

my_function(my_callback_function)

# 執行結果:
我正在執行 my_function() 函式。
我正在執行 my_callback_function() 函式。

與觀察者模式的差異:
- 觀察者模式是一種 軟體設計模式，它 允許一個物件（稱為主題）註冊其他物件（稱為觀察者），並在狀態發生變化時通知觀察者。
- callback 與觀察者模式的主要差異如下：
  - callback 是一種函式，而觀察者模式是一種 軟體設計模式。
  - callback 通常用來處理事件，而觀察者模式 通常用來在狀態發生變化時通知其他物件。
  - callback 的實現方式更加簡單，而觀察者模式的實現方式則相較之下更加複雜。
以下是一個使用觀察者模式的範例： (對比 callback 的差異)
- 這個範例中，Subject 類別可以註冊 Observer 類別的物件。Observer 類別的物件可以使用 update() 函式 來處理 Subject 類別的事件。
- 在這個範例中，subject.notify_observers() 函式 會呼叫所有註冊的 Observer 類別的物件的 update() 函式。

class Subject:
    def __init__(self):
        self.observers = []

    def register_observer(self, observer):
        self.observers.append(observer)

    def notify_observers(self):
        for observer in self.observers:
            observer.update(self)

class Observer:
    def update(self, subject):
        print("我正在執行 update() 函式。")

subject = Subject()
observer1 = Observer()
observer2 = Observer()

subject.register_observer(observer1)
subject.register_observer(observer2)

subject.notify_observers()

# 執行結果:
我正在執行 update() 函式。
我正在執行 update() 函式。

以上可知，其實 callback 和 Observer Pattern 看似相似，其實兩者互不相關，前者是 python 的一個內建函式，後者則是程式書寫的一種設計架構，當然 你也可以直接把 Observer 直接使用 callback 來實現，如下面這個範例，範例中，Subject 是被觀察的對象，Observer 是觀察者。當 Subject 的狀態發生變化時，它會通知所有已註冊的觀察者。而當 Subject 調用 notify 方法時，它會 調用所有已註冊的回調函數(callback function)。

class Subject:
    def __init__(self):
        self.callbacks = []

    def register_callback(self, callback):
        self.callbacks.append(callback)

    def notify(self, *args, **kwargs):
        for callback in self.callbacks:
            callback(*args, **kwargs)

class Observer:
    def __init__(self, name):
        self.name = name

    def __call__(self, *args, **kwargs):
        print(f"{self.name} received: {args}, {kwargs}")

subject = Subject()
observer1 = Observer("Observer 1")
observer2 = Observer("Observer 2")

subject.register_callback(observer1)
subject.register_callback(observer2)

subject.notify("Hello World!")

# 執行結果:
Observer 1 received: ('Hello World!',), {}
Observer 2 received: ('Hello World!',), {}

總結

Oberver Pattern 的優缺點優點:
- 優點:
  - 解耦合：觀察者模式將主題和觀察者解耦合，使得兩者之間的關係是鬆散的。
  - 擴展性：觀察者模式可以很容易地添加或刪除觀察者。
  - 靈活性：觀察者模式可以用於各種不同的場景。
- 缺點:
  - 可能會導致過度複雜：如果觀察者的數量過多，可能會導致系統過於複雜。
  - 可能會降低性能：如果觀察者數量過多，主題在通知所有觀察者時可能會降低性能。
Oberver Pattern 的是用場景:
- 當有多對象，想實現一次通知的廣播時，例如有一個新聞頻道，很多人訂閱，當新聞頻道有最新的新聞進來，就廣播通知所有訂閱戶
- 當一個對象的改變需要同時改變其他對象時，這很像事件訂閱的用法
總結:
- Oberver Pattern 通過定義 Subject 和 Observer 之間的一對多依賴關係,來達到當 Subject 狀態改變時,通知所有的 Observer,使它們能夠自動更新狀態。這種模式可以降低耦合度並提高擴展性。

參考資料來源

github.com/keijumt/design-pattern
github.com/mrmenand/Py_transaction
github.com/jtortorelli/head-first-design-patterns-python