Python Observer Pattern

The Observer Pattern is a behavioral design pattern that defines a one-to-many dependency relationship, allowing multiple observer objects to simultaneously listen to a certain subject object. When the subject object's state changes, all of its observers receive a notification and automatically update.

Observer Pattern in Daily Life

Imagine a newspaper subscription scenario:

Newspaper office(Subject) publishes news
Subscribers(Observer) subscribe to the newspaper
When a new newspaper is published, all subscribers automatically receive the newspaper
Subscribers can cancel their subscription at any time

This "publish-subscribe" mechanism is a typical application of the Observer Pattern.

Core Components of the Observer Pattern

The Observer Pattern contains four key roles:

1. Subject

Maintains a list of observers
Provides methods to register and unregister observers
Notifies all observers of state changes

2. Concrete Subject

Stores concrete state information
Notifies observers when state changes

3. Observer

Defines an update interface to receive subject notifications

4. Concrete Observer

Implements the update interface to stay consistent with the subject state

Implementation of the Observer Pattern

Let's understand the implementation of the Observer Pattern through a complete example.

Basic Implementation

Example

from abc import ABC, abstractmethod
from typing import List

# Observer interface
class Observer(ABC):
@abstractmethod
def update(self, message: str):
"""Observer update method"""
pass

# Subject interface
class Subject(ABC):
@abstractmethod
def register_observer(self, observer: Observer):
"""Register observer"""
pass

@abstractmethod
def remove_observer(self, observer: Observer):
"""Remove observer"""
pass

@abstractmethod
def notify_observers(self):
"""Notify all observers"""
pass

# Concrete Subject - News Publisher
class NewsPublisher(Subject):
def __init__(self):
self._observers: List[Observer] = []
self._latest_news = ""

def register_observer(self, observer: Observer):
"""Register observer"""
if observer not in self._observers:
self._observers.append(observer)
print(f"Observer {observer.__class__.__name__} registered successfully")

def remove_observer(self, observer: Observer):
"""Remove observer"""
if observer in self._observers:
self._observers.remove(observer)
print(f"Observer {observer.__class__.__name__} removed successfully")

def notify_observers(self):
"""Notify all observers"""
print(f"\nPublish new message: {self._latest_news}")
for observer in self._observers:
observer.update(self._latest_news)

def publish_news(self, news: str):
"""Publish news"""
self._latest_news = news
self.notify_observers()

def get_observer_count(self) -> int:
"""Get observer count"""
return len(self._observers)

# Concrete Observer - Email Subscriber
class EmailSubscriber(Observer):
def __init__(self, name: str):
self.name = name

def update(self, message: str):
"""Receive update and send email"""
print(f"[Email] {self.name} received message: {message}")

# Concrete Observer - SMS Subscriber
class SMSSubscriber(Observer):
def __init__(self, name: str, phone: str):
self.name = name
self.phone = phone

def update(self, message: str):
"""Receive update and send SMS"""
print(f"[SMS] {self.phone} - {self.name} received message: {message}")

# Concrete Observer - App Push Subscriber
class AppPushSubscriber(Observer):
def __init__(self, username: str):
self.username = username

def update(self, message: str):
"""Receive update and send app push notification"""
print(f"[App Push] User {self.username} received push: {message}")

Usage Example

Example

def demo_basic_observer():
"""Basic Observer Pattern demonstration"""
print("=== Basic Observer Pattern Demonstration ===\n")

# Create news publisher
publisher = NewsPublisher()

# Create different types of subscribers
email_user = EmailSubscriber("Zhang San")
sms_user = SMSSubscriber("Li Si", "13800138000")
app_user = AppPushSubscriber("Wang Wu")

# Register subscribers
publisher.register_observer(email_user)
publisher.register_observer(sms_user)
publisher.register_observer(app_user)

print(f"\nCurrent subscriber count: {publisher.get_observer_count()}")

# Publish news
publisher.publish_news("Python 3.12 officially released!")

# Remove a subscriber
publisher.remove_observer(sms_user)

print(f"\nSubscriber count after removal: {publisher.get_observer_count()}")

# Publish news again
publisher.publish_news("The Observer Pattern learning guide is online!")

# Run the demonstration
if __name__ == "__main__":
demo_basic_observer()

Output:

=== 观察者模式基础演示 ===

观察者 EmailSubscriber 注册成功
观察者 SMSSubscriber 注册成功
观察者 AppPushSubscriber 注册成功

当前订阅者数量: 3

发布新消息: Python 3.12 正式发布！
[邮件] 张三 收到消息: Python 3.12 正式发布！
[短信] 13800138000 - 李四 收到消息: Python 3.12 正式发布！
[应用推送] 用户 王五 收到推送: Python 3.12 正式发布！
观察者 SMSSubscriber 移除成功

移除后订阅者数量: 2

发布新消息: 观察者模式学习指南上线！
[邮件] 张三 收到消息: 观察者模式学习指南上线！
[应用推送] 用户 王五 收到推送: 观察者模式学习指南上线！

Advanced Application: Stock Price Monitoring System

Let's deepen our understanding through a more practical example.

Example

# Stock monitoring system example
class Stock:
"""Stock class"""
def __init__(self, symbol: str, price: float):
self.symbol = symbol # Stock code
self._price = price # Current price
self._history = [price] # Price history

@property
def price(self) -> float:
return self._price

@price.setter
def price(self, value: float):
"""Set price and record history"""
old_price = self._price
self._price = value
self._history.append(value)
return old_price, value

def get_price_change(self) -> float:
"""Get price change percentage"""
if len(self._history) < 2:
return 0.0
return ((self._history[-1] - self._history[-2]) / self._history[-2]) * 100

class StockMarket(Subject):
"""Stock market - Concrete Subject"""
def __init__(self):
self._observers: List[Observer] = []
self._stocks = {} # Stock dictionary: {code: Stock object}

def register_observer(self, observer: Observer):
if observer not in self._observers:
self._observers.append(observer)

def remove_observer(self, observer: Observer):
if observer in self._observers:
self._observers.remove(observer)

def notify_observers(self, stock: Stock):
"""Notify observers of a specific stock's change"""
change = stock.get_price_change()
message = f"{stock.symbol}: ${stock.price:.2f} ({change:+.2f}%)"

for observer in self._observers:
observer.update(message)

def add_stock(self, symbol: str, initial_price: float):
"""Add a stock to the market"""
self._stocks[symbol] = Stock(symbol, initial_price)

def update_stock_price(self, symbol: str, new_price: float):
"""Update stock price and notify observers"""
if symbol in self._stocks:
stock = self._stocks[symbol]
old_price, current_price = stock.price, new_price
stock.price = new_price

# Only notify when the price actually changes
if old_price != current_price:
self.notify_observers(stock)

class Investor(Observer):
"""Investor - Concrete Observer"""
def __init__(self, name: str, interest_stocks: List[str] = None):
self.name = name
self.interest_stocks = interest_stocks or []
self.notifications = []

def update(self, message: str):
"""Receive stock updates"""
stock_symbol = message.split(":")[0]

# Only pay attention to stocks of interest
if not self.interest_stocks or stock_symbol in self.interest_stocks:
self.notifications.append(message)
print(f"Investor {self.name} received: {message}")

def show_notifications(self):
"""Display all notifications"""
print(f"\n=== {self.name}'s notification records ===")
for notification in self.notifications:
print(f" - {notification}")

def demo_stock_market():
"""Stock market demonstration"""
print("\n=== Stock Market Monitoring System Demonstration ===\n")

# Create stock market
market = StockMarket()

# Add stocks
market.add_stock("AAPL", 150.0)
market.add_stock("GOOGL", 2800.0)
market.add_stock("TSLA", 250.0)

# Create investors
investor1 = Investor("Technology Investor", ["AAPL", "GOOGL"])
investor2 = Investor("New Energy Investor", ["TSLA"])
investor3 = Investor("Whole Market Investor")

# Register investors
market.register_observer(investor1)
market.register_observer(investor2)
market.register_observer(investor3)

# Simulate stock price changes
print("Start simulating stock trading...")
market.update_stock_price("AAPL", 152.5) # AAPL rises
market.update_stock_price("TSLA", 245.0) # TSLA falls
market.update_stock_price("GOOGL", 2820.0) # GOOGL rises
market.update_stock_price("AAPL", 151.0) # AAPL falls slightly

# Display investor notification records
investor1.show_notifications()
investor2.show_notifications()
investor3.show_notifications()

# Run stock market demonstration
demo_stock_market()

Advantages and Disadvantages of the Observer Pattern

Advantages

Advantages	Description
Loose Coupling	The dependency between subject and observers is loose, allowing them to change independently
Extensibility	New observers can be easily added without modifying the subject code
Broadcast Communication	Supports one-to-many communication; messages can be broadcast to multiple objects
Open/Closed Principle	Open for extension, closed for modification

Disadvantages

Disadvantages	Description	Solution
Memory Leak	Observers forgetting to unregister may cause memory leaks	Use weak references or ensure timely unregistration
Update Order	The update order of observers is uncertain	If order is needed, a priority mechanism can be added
Performance Issues	When there are many observers, notifications may affect performance	Use asynchronous notifications or batch updates

Practical Application Scenarios

1. GUI Event Handling

Example

# Pseudocode example
button.on_click(handler1) # Register click event observer
button.on_click(handler2) # Register another observer
# When button is clicked, all registered handlers will be called

2. Message Queue System

Producer publishes message
Multiple consumers subscribe and process messages

3. Data Monitoring System

Data source state changes
Multiple monitoring services receive notifications and take action

4. Game Development

Game event system
UI elements respond to game state changes

Best Practices and Considerations

1. Use Weak References to Avoid Memory Leaks

Example

import weakref

class SafeSubject(Subject):
def __init__(self):
self._observers = weakref.WeakSet() # Use weak reference collection

def register_observer(self, observer: Observer):
self._observers.add(observer)

2. Asynchronous Notification to Improve Performance

Example

import asyncio

class AsyncObserver(ABC):
@abstractmethod
async def update_async(self, message: str):
pass

class AsyncSubject:
async def notify_observers_async(self):
tasks = [observer.update_async(self._message)
for observer in self._observers]
await asyncio.gather(*tasks)

3. Add Error Handling

Example

def safe_notify(self):
for observer in self._observers.copy(): # Use a copy to avoid modification issues
try:
observer.update(self._message)
except Exception as e:
print(f"Observer {observer} update failed: {e}")

Summary

The Observer pattern is a very important design pattern in Python. It provides an elegant way to achieve loose coupling between objects. After reading this article, you should be able to:

Understand the core concepts and applicable scenarios of the Observer pattern
Master the basic implementation methods of the Observer pattern
Apply the Observer pattern to solve problems in real projects
Avoid common pitfalls of the Observer pattern

Other Extensions