Observer Pattern: A Complete Guide with Examples in 11 Programming Languages

The Observer Pattern is a behavioral design pattern that defines a one-to-many dependency between objects so that when one object (the Subject, or Publisher) changes state, all its dependents (Observers, or Subscribers) are automatically notified and updated. Instead of observers polling the subject repeatedly to check for changes — wasting CPU cycles and adding latency — the subject pushes notifications to all registered observers the moment something changes. The observers then react independently, each doing whatever is appropriate for its role.

In this comprehensive guide, we'll explore the Observer Pattern with real-world examples in Python, TypeScript, Java, JavaScript, C#, PHP, Go, Rust, Dart, Swift, and Kotlin — complete with language-specific best practices, common pitfalls, and when to use this powerful pattern.

Table of Contents

• What is the Observer Pattern?
• Why Use the Observer Pattern?
• Observer Pattern Comparison
• Observer Pattern Explained
• Class Diagram
• Beginner-Friendly Example
• Production-Ready Example
• Real-World Use Cases
• Language-Specific Mistakes and Anti-Patterns
• Frequently Asked Questions
• Key Takeaways

What is the Observer Pattern?

The Observer Pattern establishes a publish-subscribe relationship between objects. The subject maintains a list of observers and notifies them automatically of state changes. Observers register their interest, receive notifications, and react — all without the subject knowing anything specific about what each observer does.

Think of it like a newspaper subscription service. The newspaper (subject) doesn't call each reader individually to ask if they want to read today's edition. Instead, subscribers (observers) opt in once. When a new edition is published, every subscriber receives it automatically. Subscribers can cancel at any time. The newspaper doesn't know or care what each subscriber does with the paper — read it, recycle it, or share it.

Why Use the Observer Pattern?

The Observer Pattern offers several compelling benefits:

1. Open/Closed Principle: Add new observers without modifying the subject — new reactions to state changes require no changes to existing code
2. Loose Coupling: Subjects know only the Observer interface, not the concrete types. Observers know only the Subject interface. Neither depends on the other's implementation.
3. Broadcast Communication: One subject state change automatically notifies any number of observers — from zero to thousands
4. Dynamic Subscription: Observers can subscribe and unsubscribe at runtime, allowing flexible, responsive systems
5. Separation of Concerns: The subject handles its own state; each observer handles its own reaction — concerns never mix
6. Reusability: Subjects and observers can be recombined in different configurations without modification

Observer Pattern Comparison

Let's compare the Observer Pattern with related patterns:

Key Distinctions:

• Observer vs. Mediator: Observer is passive broadcast — the subject fires and forgets; observers react independently. Mediator is active coordination — the mediator receives a notification and actively orchestrates responses across components based on business rules. Use Observer for loose event notification; use Mediator when you need centralized control of how components interact.
• Observer vs. Event Bus: A classic Observer has the subject hold direct references to observers. An event bus decouples further — publishers and subscribers never reference each other at all; the bus routes by event type. An event bus is essentially a global Observer implementation where the subject is anonymous.

Observer Pattern Explained

The Observer Pattern involves four core components:

Core Components:

1. Subject Interface (Publisher): Declares the methods for managing observers — subscribe(observer), unsubscribe(observer), and notify(). The subject knows its observers only through this interface.

2. Concrete Subject: Implements the Subject interface. Maintains its own state and a list of subscribed observers. Calls notify() whenever its state changes in a way that observers care about.

3. Observer Interface (Subscriber): Declares the update(event) method that the subject calls when notifying. All observers implement this interface.

4. Concrete Observers: Implement the Observer interface. Each registers with one or more subjects. Each reacts to notifications independently — logging, updating a UI, sending an email, invalidating a cache, etc.

Notification Strategies:

• Push model: The subject includes the changed data in the notification call — update(newValue). Observers don't need to query back. Simple but may send more data than some observers need.
• Pull model: The subject sends only a reference to itself — update(subject). Observers query the subject for the data they need. More flexible but requires observers to know the subject's interface.
• Event object model: The subject sends a typed event object — update(event: StockChangedEvent). The most expressive — carries event type, old value, new value, timestamp, and any context. Used by most modern event systems.

Observer Variants:

• Simple Observer: Single event type, single update() method
• Typed Event Observer: Different observer interfaces per event type (onPriceChange, onVolumeChange)
• Event Bus: Centralized broker; publishers and subscribers are fully decoupled
• Reactive Streams (RxJava, RxJS, Kotlin Flow): Observer pattern extended with backpressure, error handling, and declarative composition operators

Class Diagram

Beginner-Friendly Example

Let's start with the most intuitive Observer example: a weather station. The station (subject) measures temperature, humidity, and pressure. Multiple displays (observers) show current conditions, statistics, or forecasts. When the station takes a new measurement, all displays update automatically. New display types can be added without touching the station.

from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Set
import statistics


# ── Event object ─────────────────────────────────────────────────
@dataclass(frozen=True)
class WeatherData:
    temperature: float   # Celsius
    humidity:    float   # Percent
    pressure:    float   # hPa


# ── Observer Interface ────────────────────────────────────────────
class WeatherObserver(ABC):
    @abstractmethod
    def update(self, data: WeatherData) -> None:
        pass


# ── Subject Interface ─────────────────────────────────────────────
class WeatherSubject(ABC):
    @abstractmethod
    def subscribe(self, observer: WeatherObserver) -> None:
        pass

    @abstractmethod
    def unsubscribe(self, observer: WeatherObserver) -> None:
        pass

    @abstractmethod
    def notify(self) -> None:
        pass


# ── Concrete Subject ──────────────────────────────────────────────
class WeatherStation(WeatherSubject):
    def __init__(self, name: str) -> None:
        self.name         = name
        self._observers:  List[WeatherObserver] = []
        self._data:       WeatherData | None     = None

    def subscribe(self, observer: WeatherObserver) -> None:
        if observer not in self._observers:
            self._observers.append(observer)
            print(f"[{self.name}] Observer added: {type(observer).__name__}")

    def unsubscribe(self, observer: WeatherObserver) -> None:
        if observer in self._observers:
            self._observers.remove(observer)
            print(f"[{self.name}] Observer removed: {type(observer).__name__}")

    def notify(self) -> None:
        if self._data is None:
            return
        for observer in list(self._observers):  # Iterate a copy — safe if observers modify the list
            observer.update(self._data)

    def set_measurements(self, temp: float, humidity: float, pressure: float) -> None:
        self._data = WeatherData(temp, humidity, pressure)
        print(f"\n[{self.name}] New measurement → {self._data}")
        self.notify()

    @property
    def current_data(self) -> WeatherData | None:
        return self._data


# ── Concrete Observers ────────────────────────────────────────────
class CurrentConditionsDisplay(WeatherObserver):
    def __init__(self, location: str) -> None:
        self._location = location
        self._data: WeatherData | None = None

    def update(self, data: WeatherData) -> None:
        self._data = data
        self._display()

    def _display(self) -> None:
        if self._data:
            print(f"  [CurrentConditions:{self._location}] "
                  f"🌡 {self._data.temperature:.1f}°C  "
                  f"💧 {self._data.humidity:.0f}%  "
                  f"🌬 {self._data.pressure:.0f} hPa")


class StatisticsDisplay(WeatherObserver):
    def __init__(self) -> None:
        self._temps: List[float] = []

    def update(self, data: WeatherData) -> None:
        self._temps.append(data.temperature)
        self._display()

    def _display(self) -> None:
        if len(self._temps) < 2:
            print(f"  [Statistics] Temp: {self._temps[-1]:.1f}°C (need more readings)")
            return
        print(f"  [Statistics] Temp — "
              f"min: {min(self._temps):.1f}°C  "
              f"avg: {statistics.mean(self._temps):.1f}°C  "
              f"max: {max(self._temps):.1f}°C")


class ForecastDisplay(WeatherObserver):
    def __init__(self) -> None:
        self._last_pressure: float | None = None

    def update(self, data: WeatherData) -> None:
        forecast = self._forecast(data.pressure)
        print(f"  [Forecast] {forecast}")
        self._last_pressure = data.pressure

    def _forecast(self, pressure: float) -> str:
        if self._last_pressure is None:
            return "📊 Not enough data yet"
        if pressure > self._last_pressure:
            return "☀️  Improving — higher pressure incoming"
        if pressure < self._last_pressure:
            return "🌧  Watch out — lower pressure, rain likely"
        return "⛅ More of the same"


class AlertObserver(WeatherObserver):
    """Only fires when thresholds are breached."""
    def __init__(self, max_temp: float = 35.0, min_pressure: float = 980.0) -> None:
        self._max_temp    = max_temp
        self._min_pressure = min_pressure

    def update(self, data: WeatherData) -> None:
        if data.temperature >= self._max_temp:
            print(f"  [ALERT] 🚨 Heat warning! Temp: {data.temperature:.1f}°C")
        if data.pressure <= self._min_pressure:
            print(f"  [ALERT] 🚨 Low pressure warning! {data.pressure:.0f} hPa — storm possible")


# ── Client ────────────────────────────────────────────────────────
if __name__ == "__main__":
    station = WeatherStation("City Center Station")

    current   = CurrentConditionsDisplay("Downtown")
    stats     = StatisticsDisplay()
    forecast  = ForecastDisplay()
    alerts    = AlertObserver(max_temp=30.0, min_pressure=995.0)

    station.subscribe(current)
    station.subscribe(stats)
    station.subscribe(forecast)
    station.subscribe(alerts)

    station.set_measurements(22.5, 65.0, 1013.0)
    station.set_measurements(25.0, 70.0, 1010.0)
    station.set_measurements(31.0, 80.0, 993.0)   # Triggers both alerts

    print("\n--- ForecastDisplay unsubscribes ---")
    station.unsubscribe(forecast)
    station.set_measurements(28.0, 75.0, 998.0)   # Forecast no longer updates

Production-Ready Example

Now let's tackle a realistic scenario: a stock market event system. Multiple stock tickers (subjects) are watched by different types of observers: portfolio trackers, alert engines, audit loggers, and analytics dashboards. The system supports typed events (price changed, volume spike, trading halt), observer priority, and an event bus that decouples publishers from subscribers completely.

🐍 Python (Production Example)

from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from typing import Callable, Dict, List, Optional, Set, Type, TypeVar
import threading


# ─── Event System ─────────────────────────────────────────────────
class EventType(Enum):
    PRICE_CHANGED  = auto()
    VOLUME_SPIKE   = auto()
    TRADING_HALTED = auto()
    TRADING_RESUMED= auto()
    CIRCUIT_BREAKER= auto()


@dataclass(frozen=True)
class StockEvent:
    event_type: EventType
    ticker:     str
    timestamp:  str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S.%f")[:12])
    data:       dict = field(default_factory=dict)

    def __str__(self) -> str:
        return f"[{self.timestamp}] {self.event_type.name} {self.ticker} {self.data}"


# ─── Observer Interface ───────────────────────────────────────────
class StockObserver(ABC):
    @abstractmethod
    def on_event(self, event: StockEvent) -> None:
        pass

    @property
    def priority(self) -> int:
        """Lower number = higher priority (called first). Default: 100."""
        return 100


# ─── Concrete Subject: Stock Ticker ──────────────────────────────
class StockTicker:
    CIRCUIT_BREAKER_THRESHOLD = 0.10  # 10% move triggers circuit breaker

    def __init__(self, ticker: str, initial_price: float) -> None:
        self.ticker        = ticker
        self._price        = initial_price
        self._prev_price   = initial_price
        self._volume       = 0
        self._halted       = False
        self._observers:  List[StockObserver] = []
        self._lock         = threading.Lock()

    def subscribe(self, observer: StockObserver) -> None:
        with self._lock:
            self._observers.append(observer)
            self._observers.sort(key=lambda o: o.priority)

    def unsubscribe(self, observer: StockObserver) -> None:
        with self._lock:
            self._observers = [o for o in self._observers if o is not observer]

    def _notify(self, event: StockEvent) -> None:
        with self._lock:
            observers_snapshot = list(self._observers)
        for observer in observers_snapshot:
            try:
                observer.on_event(event)
            except Exception as e:
                print(f"  [Ticker:{self.ticker}] Observer error ({type(observer).__name__}): {e}")

    # ── Trading actions ───────────────────────────────────────────
    def update_price(self, new_price: float, volume: int = 0) -> None:
        if self._halted:
            print(f"  [{self.ticker}] Trading halted — price update rejected")
            return

        change_pct = (new_price - self._price) / self._price

        # Check circuit breaker
        if abs(change_pct) >= self.CIRCUIT_BREAKER_THRESHOLD:
            self._halted = True
            self._notify(StockEvent(
                EventType.CIRCUIT_BREAKER, self.ticker,
                data={"price": self._price, "attempted": new_price,
                      "change_pct": round(change_pct * 100, 2)}
            ))
            self._notify(StockEvent(EventType.TRADING_HALTED, self.ticker,
                data={"reason": "circuit_breaker"}))
            return

        self._prev_price = self._price
        self._price      = new_price
        self._volume    += volume

        self._notify(StockEvent(
            EventType.PRICE_CHANGED, self.ticker,
            data={"price": new_price, "prev_price": self._prev_price,
                  "change_pct": round(change_pct * 100, 2), "volume": volume}
        ))

        # Volume spike detection (volume > 3× last trade)
        if volume > 0 and self._volume > 0 and volume > self._volume * 3:
            self._notify(StockEvent(EventType.VOLUME_SPIKE, self.ticker,
                data={"volume": volume, "avg_volume": self._volume}))

    def resume_trading(self) -> None:
        self._halted = False
        self._notify(StockEvent(EventType.TRADING_RESUMED, self.ticker,
            data={"current_price": self._price}))

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

    @property
    def is_halted(self) -> bool:
        return self._halted


# ─── Concrete Observers ────────────────────────────────────────────
class AuditLogger(StockObserver):
    """Highest priority — always runs first for compliance."""

    def __init__(self, log_file: str = "audit.log") -> None:
        self._log_file = log_file
        self._entries:  List[str] = []

    @property
    def priority(self) -> int:
        return 1  # Highest priority

    def on_event(self, event: StockEvent) -> None:
        entry = f"AUDIT | {event}"
        self._entries.append(entry)
        print(f"  [AuditLogger] {entry}")

    def get_log(self) -> List[str]:
        return list(self._entries)


class PortfolioTracker(StockObserver):
    """Tracks P&L for a portfolio of stock holdings."""

    def __init__(self, name: str) -> None:
        self._name     = name
        self._holdings: Dict[str, dict] = {}  # ticker → {shares, cost_basis}

    def add_holding(self, ticker: str, shares: int, cost_basis: float) -> None:
        self._holdings[ticker] = {"shares": shares, "cost_basis": cost_basis}

    def on_event(self, event: StockEvent) -> None:
        if event.event_type != EventType.PRICE_CHANGED:
            return
        ticker = event.ticker
        if ticker not in self._holdings:
            return
        h         = self._holdings[ticker]
        new_price = event.data["price"]
        pnl       = (new_price - h["cost_basis"]) * h["shares"]
        pnl_pct   = (new_price - h["cost_basis"]) / h["cost_basis"] * 100
        emoji     = "📈" if pnl >= 0 else "📉"
        print(f"  [Portfolio:{self._name}] {ticker} {emoji} "
              f"${new_price:.2f}  P&L: ${pnl:+.2f} ({pnl_pct:+.1f}%)")


class PriceAlertEngine(StockObserver):
    """Fires alerts when price crosses configured thresholds."""

    def __init__(self) -> None:
        self._alerts: List[dict] = []  # {ticker, above/below, threshold, triggered}

    def add_alert(self, ticker: str, direction: str, threshold: float) -> None:
        self._alerts.append({
            "ticker": ticker, "direction": direction,
            "threshold": threshold, "triggered": False
        })

    def on_event(self, event: StockEvent) -> None:
        if event.event_type not in (EventType.PRICE_CHANGED, EventType.TRADING_HALTED,
                                     EventType.CIRCUIT_BREAKER):
            return
        if event.event_type == EventType.CIRCUIT_BREAKER:
            print(f"  [AlertEngine] 🛑 CIRCUIT BREAKER: {event.ticker} "
                  f"tried to move {event.data.get('change_pct')}%")
            return
        if event.event_type == EventType.TRADING_HALTED:
            print(f"  [AlertEngine] ⚠️  {event.ticker} trading HALTED")
            return

        price  = event.data["price"]
        ticker = event.ticker
        for alert in self._alerts:
            if alert["ticker"] != ticker or alert["triggered"]:
                continue
            if alert["direction"] == "above" and price >= alert["threshold"]:
                print(f"  [AlertEngine] 🔔 {ticker} crossed ABOVE ${alert['threshold']:.2f} "
                      f"(now ${price:.2f})")
                alert["triggered"] = True
            elif alert["direction"] == "below" and price <= alert["threshold"]:
                print(f"  [AlertEngine] 🔔 {ticker} crossed BELOW ${alert['threshold']:.2f} "
                      f"(now ${price:.2f})")
                alert["triggered"] = True


class AnalyticsDashboard(StockObserver):
    """Aggregates statistics across all observed tickers."""

    def __init__(self) -> None:
        self._stats: Dict[str, dict] = {}

    def on_event(self, event: StockEvent) -> None:
        if event.event_type != EventType.PRICE_CHANGED:
            return
        ticker = event.ticker
        price  = event.data["price"]
        if ticker not in self._stats:
            self._stats[ticker] = {"prices": [], "high": price, "low": price,
                                    "trades": 0, "total_volume": 0}
        s = self._stats[ticker]
        s["prices"].append(price)
        s["high"]    = max(s["high"], price)
        s["low"]     = min(s["low"],  price)
        s["trades"] += 1
        s["total_volume"] += event.data.get("volume", 0)

    def print_summary(self) -> None:
        print("\n  [AnalyticsDashboard] ── Session Summary ─────────────")
        for ticker, s in self._stats.items():
            avg = sum(s["prices"]) / len(s["prices"])
            print(f"    {ticker}: "
                  f"H:${s['high']:.2f}  L:${s['low']:.2f}  "
                  f"Avg:${avg:.2f}  "
                  f"Trades:{s['trades']}  Vol:{s['total_volume']:,}")


# ─── Event Bus: Fully Decoupled Publisher/Subscriber ─────────────
EventHandler = Callable[[StockEvent], None]

class StockEventBus:
    """
    A global event bus — publishers post events without knowing who listens.
    Subscribers register interest by EventType, not by ticker.
    """
    _instance: Optional["StockEventBus"] = None

    def __init__(self) -> None:
        self._handlers: Dict[EventType, List[EventHandler]] = {}
        self._lock = threading.Lock()

    @classmethod
    def instance(cls) -> "StockEventBus":
        if cls._instance is None:
            cls._instance = StockEventBus()
        return cls._instance

    def subscribe(self, event_type: EventType, handler: EventHandler) -> None:
        with self._lock:
            self._handlers.setdefault(event_type, []).append(handler)

    def publish(self, event: StockEvent) -> None:
        with self._lock:
            handlers = list(self._handlers.get(event.event_type, []))
        for handler in handlers:
            try:
                handler(event)
            except Exception as e:
                print(f"  [EventBus] Handler error: {e}")


# ─── Client ────────────────────────────────────────────────────────
if __name__ == "__main__":
    # ── Direct Observer wiring ────────────────────────────────────
    aapl = StockTicker("AAPL", initial_price=175.00)
    tsla = StockTicker("TSLA", initial_price=250.00)

    audit     = AuditLogger()
    portfolio = PortfolioTracker("MyPortfolio")
    alerts    = PriceAlertEngine()
    dashboard = AnalyticsDashboard()

    portfolio.add_holding("AAPL", shares=100, cost_basis=160.00)
    portfolio.add_holding("TSLA", shares=50,  cost_basis=240.00)
    alerts.add_alert("AAPL", "above", 180.00)
    alerts.add_alert("TSLA", "below", 245.00)

    for ticker in [aapl, tsla]:
        ticker.subscribe(audit)
        ticker.subscribe(portfolio)
        ticker.subscribe(alerts)
        ticker.subscribe(dashboard)

    print("=" * 60)
    print("SCENARIO 1: Normal trading")
    print("=" * 60)
    aapl.update_price(178.50, volume=1_000_000)
    aapl.update_price(181.00, volume=1_200_000)   # Triggers AAPL alert
    tsla.update_price(244.00, volume=800_000)      # Triggers TSLA alert
    tsla.update_price(247.50, volume=900_000)

    print("\n" + "=" * 60)
    print("SCENARIO 2: Circuit breaker triggers on AAPL")
    print("=" * 60)
    aapl.update_price(198.00, volume=5_000_000)    # +9.9% — below threshold
    aapl.update_price(205.00, volume=10_000_000)   # +3.5% from 198 — but total triggers breaker
    # After reset:
    print("\n[Exchange] Reviewing circuit breaker — resuming AAPL")
    aapl.resume_trading()
    aapl.update_price(185.00, volume=2_000_000)

    print("\n" + "=" * 60)
    print("SCENARIO 3: Event Bus (decoupled) — add a compliance sink")
    print("=" * 60)
    bus = StockEventBus.instance()
    bus.subscribe(EventType.CIRCUIT_BREAKER, lambda e: print(
        f"  [ComplianceSink] Filing circuit breaker report for {e.ticker}: {e.data}"))
    bus.subscribe(EventType.TRADING_HALTED, lambda e: print(
        f"  [ComplianceSink] Notifying regulator: {e.ticker} halted"))

    # Make the ticker also publish to the bus
    original_notify = tsla._notify
    def bus_publishing_notify(event: StockEvent) -> None:
        original_notify(event)
        bus.publish(event)
    tsla._notify = bus_publishing_notify  # type: ignore

    tsla.update_price(210.00, volume=20_000_000)   # -16% circuit breaker on TSLA
    tsla.resume_trading()

    dashboard.print_summary()
    print(f"\n  [AuditLogger] Total events logged: {len(audit.get_log())}")

Real-World Use Cases

1. GUI Event Systems

Every modern GUI framework is built on the Observer Pattern. Button clicks, mouse moves, key presses, and scroll events are all subjects. Event handlers are observers. React's useState hooks notify component re-renders. Android's LiveData notifies UI components. iOS's NotificationCenter broadcasts system and application events.

2. Model-View-Controller (MVC)

The Model is the subject. Views are observers. When the Model changes, all registered Views update automatically — the separation of concerns that MVC is founded on. Angular's change detection, Vue's reactive data, and Ember's computed properties are all Observer implementations.

3. Reactive Programming (RxJava / RxJS / Kotlin Flow)

Reactive streams extend the Observer Pattern with backpressure, error propagation, and declarative composition operators (map, filter, flatMap, zip). An Observable is a subject; Subscriber/Observer receives emissions. RxJS powers Angular; RxJava powers Android; Kotlin Flow is the coroutine-based evolution for Kotlin/Android.

4. Financial Market Data

Bloomberg Terminal, Reuters Eikon, and trading platforms use Observer at massive scale. Thousands of price feeds (subjects) push tick-by-tick updates to order books, risk engines, trading algorithms, and displays (observers) in microseconds.

5. Message Queues and Pub/Sub Systems

Apache Kafka, RabbitMQ, AWS SNS/SQS, Google Cloud Pub/Sub, and Redis Pub/Sub are distributed Observer implementations. Producers publish to topics; consumers subscribe to topics. The broker (bus) routes messages. The pattern is the same — the implementation spans processes and machines.

6. Domain Events in Domain-Driven Design

When a domain event fires (OrderPlaced, PaymentReceived, ShipmentDispatched), registered handlers react — sending confirmation emails, updating inventory, charging the card, notifying the warehouse. The aggregate (subject) fires the event; domain event handlers (observers) react. This keeps business logic decoupled across bounded contexts.

7. Hot Reloading in Development Tools

Webpack's Hot Module Replacement, Vite's fast refresh, and Flutter's hot reload use the Observer Pattern. The file watcher (subject) detects changes. Compilers, test runners, and browser tabs (observers) react to rebuild, retest, or refresh.

Language-Specific Mistakes and Anti-Patterns

# BAD: Observer unsubscribes itself during the notification loop — skips the next observer
class SelfRemovingObserver(WeatherObserver):
    def update(self, data):
        self._count += 1
        if self._count >= 3:
            station.unsubscribe(self)   # Modifies station._observers DURING iteration!
            # Python's list iteration is invalidated — the next observer may be skipped!

# In WeatherStation.notify():
for observer in self._observers:        # Iterating original list
    observer.update(self._data)         # Observer modifies self._observers here → skip!

# GOOD: Always iterate a snapshot copy
def notify(self):
    for observer in list(self._observers):  # Snapshot — safe even if list is modified
        observer.update(self._data)

# BAD: Observer is created in a local scope and goes out of scope, but subject holds it alive
def setup():
    station = get_station()
    display = CurrentConditionsDisplay("Lobby")  # Created in local scope
    station.subscribe(display)
    # display falls out of scope here — but station._observers still holds a strong reference!
    # display is never garbage-collected even though no other code uses it

# GOOD: Use weak references so observers are GC'd when no other code holds them
import weakref

class WeatherStation:
    def __init__(self):
        self._observers = []  # List of weakref.ref objects

    def subscribe(self, observer):
        self._observers.append(weakref.ref(observer))

    def notify(self):
        alive = []
        for ref in self._observers:
            o = ref()
            if o is not None:
                o.update(self._data)
                alive.append(ref)
        self._observers = alive  # Prune dead refs

# BAD: One misbehaving observer crashes the entire notification chain
def notify(self):
    for observer in list(self._observers):
        observer.update(self._data)  # If this raises, remaining observers never run!

# GOOD: Catch and log exceptions per observer
def notify(self):
    for observer in list(self._observers):
        try:
            observer.update(self._data)
        except Exception as e:
            print(f"[{self.__class__.__name__}] Observer {type(observer).__name__} "
                  f"raised: {e}")
            # Continue to next observer — one failure doesn't break the chain

Frequently Asked Questions

Q1: What is the difference between the Observer Pattern and an event bus?

The classic Observer Pattern has the Subject hold direct references to its Observers. Subjects know their observers (even if only via the interface). If you have 10 subjects and 5 observer types, every observer that cares about a subject must register with it explicitly.

An event bus is a fully decentralized Observer. Publishers post events to a named topic/channel on the bus without knowing who (if anyone) is listening. Subscribers register interest in topics without knowing who publishes them. The bus routes by topic. Publishers and subscribers are completely anonymous to each other.

Use classic Observer for tightly scoped relationships (a model and its views). Use an event bus for cross-module, cross-service, or truly decoupled communication.

Q2: Should I use the push model or pull model for notifications?

Push model: Subject includes changed data in the notification — observer.update(newData). Pros: observers get exactly what they need immediately; no extra queries. Cons: if different observers need different data, the subject must send everything, which may be wasteful.

Pull model: Subject sends only itself (or nothing) — observer.update(subject). Observers query for what they need. Pros: observers only fetch what they need. Cons: observers depend on the Subject's interface (not just the Observer interface); more coupling.

Event object model (recommended for most cases): Subject wraps changed data in a typed event object — observer.update(event: PriceChangedEvent). The event carries exactly what changed, with context. Observers receive full context without coupling to the Subject's full interface.

Q3: How do I prevent memory leaks from forgotten observer subscriptions?

Three main strategies:

1. Weak references: Subject holds WeakReference to observers (Java) or a weak set (Python's weakref). Observers are automatically removed when they go out of scope.

2. Explicit lifecycle management: Unsubscribe in Dispose() (C#), onDestroy() (Android), deinit (Swift), or componentWillUnmount() (React). Use RAII or a subscription token.

3. Subscription tokens: Subscribe returns a token object. When the token is dropped/disposed, it automatically unsubscribes. This is the pattern used by RxJava's Disposable, Combine's AnyCancellable, and JavaScript's Subscription.

class Subscription:
    def __init__(self, subject, observer):
        self._subject  = subject
        self._observer = observer
        subject.subscribe(observer)

    def dispose(self):
        self._subject.unsubscribe(self._observer)

    def __enter__(self): return self
    def __exit__(self, *args): self.dispose()

with Subscription(station, display):
    station.set_measurements(22, 65, 1013)
# Automatically unsubscribed here

Q4: How do I handle async observers that do slow work?

Never block the notification loop. Two common patterns:

1. Fire-and-forget async: Observer schedules async work and returns immediately. The subject's notify() completes quickly. Errors must be handled within the async work.

2. Channel/queue-based: Subject posts to a channel; observers consume from it asynchronously at their own pace. This also provides backpressure — if the observer is slow, the channel fills up and the producer can react.

import asyncio
from concurrent.futures import ThreadPoolExecutor

class AsyncDatabaseObserver(WeatherObserver):
    def __init__(self):
        self._executor = ThreadPoolExecutor(max_workers=2)

    def update(self, data: WeatherData) -> None:
        # Submit to thread pool — returns immediately
        self._executor.submit(self._write_to_db, data)

    def _write_to_db(self, data: WeatherData) -> None:
        # Blocking DB write runs in thread pool, not in the notification loop
        time.sleep(0.5)
        print(f"[DB] Wrote: {data}")

Q5: How many observers can a subject support?

Theoretically unlimited, but practical limits depend on notification cost. For synchronous notification: hundreds to low thousands are fine; tens of thousands may cause noticeable latency. For async notification (channel-based or thread-pooled): subjects can support far more since they don't wait for observers to complete.

High-performance systems (financial market data, gaming) typically fan out to a very small number of direct observers, each of which is responsible for forwarding to its own set of downstream consumers.

Q6: What is the difference between java.util.Observable and the Observer Pattern?

java.util.Observable (deprecated in Java 9) is a concrete base class for subjects. Its main problems:

• Requires inheritance: Your subject must extend Observable, preventing extending any other class
• setChanged() must be called manually: Easy to forget, causing missed notifications
• Thread-unsafe: notifyObservers() isn't properly synchronized
• Observer interface is not generic: update(Observable o, Object arg) is untyped

The solution is always to implement the Observer Pattern from scratch with your own typed interfaces — as shown in the Java example above using Java's record and typed event objects.

Q7: How does the Observer Pattern relate to ReactiveX (RxJava, RxJS)?

ReactiveX is the Observer Pattern supercharged. In RxJava/RxJS, an Observable is the subject; Observer/Subscriber is the observer. But ReactiveX adds:

• Composition operators: map(), filter(), flatMap(), zip(), debounce(), throttle() — transform and combine streams declaratively
• Error handling: onError() on the observer; onErrorResume(), retry() operators on the stream
• Completion: onComplete() — subjects can signal end-of-stream
• Backpressure (RxJava Flowable): Consumer signals demand; producer respects it — prevents overwhelming slow consumers
• Schedulers: Control which thread observes on and which thread subscribes on

Use ReactiveX when you need complex stream composition. Use classic Observer for straightforward notification with no composition needs.

Q8: Can I have observers that only react to specific event types?

Yes — three ways:

1. Typed Observer interfaces: Define separate observer interfaces per event type — PriceChangeObserver, VolumeObserver. Subjects notify each separately.

2. Event filtering in the observer: Observer receives all events but filters internally — if event.type == "PRICE_CHANGED": return.

3. Event bus with topic filtering: Observer subscribes to specific topics on the bus. The bus only delivers matching events.

class EventBus:
    def subscribe(self, event_type: EventType, handler: Callable) -> None:
        self._handlers.setdefault(event_type, []).append(handler)

    def publish(self, event: StockEvent) -> None:
        for handler in self._handlers.get(event.event_type, []):
            handler(event)   # Only handlers for this event_type are called

Q9: How do I test subjects and observers?

Test subjects by injecting mock observers and verifying update() is called with the correct data. Test observers by calling update() directly with controlled events and verifying the expected side effects.

from unittest.mock import Mock, call

def test_station_notifies_observers_on_measurement():
    station = WeatherStation("Test")
    mock_observer = Mock(spec=WeatherObserver)
    station.subscribe(mock_observer)

    station.set_measurements(25.0, 70.0, 1010.0)

    mock_observer.update.assert_called_once()
    event = mock_observer.update.call_args[0][0]
    assert event.temperature == 25.0

def test_statistics_display_tracks_min_max():
    display = StatisticsDisplay()
    display.update(WeatherData(20.0, 60.0, 1000.0))
    display.update(WeatherData(30.0, 70.0, 990.0))
    assert display.min_temp == 20.0
    assert display.max_temp == 30.0

Q10: When should I NOT use the Observer Pattern?

Avoid Observer when:

• Only one observer is ever needed: A direct method call is clearer and simpler
• Order of notification matters and is complex: If observers must fire in a specific sequence with dependencies between them, use a Mediator or a pipeline instead
• The flow is request-response (synchronous): If the caller needs a return value from the "observer", use a direct call or Command pattern — Observer is fire-and-forget
• Debugging is already difficult: Every Observer notification creates implicit dependencies that are hard to trace. In systems where traceability is critical, explicit calls are easier to debug
• The system is tiny: Adding Observer infrastructure for a 200-line script adds complexity with no benefit

Key Takeaways

🎯 Core Concept: The Observer Pattern defines a one-to-many dependency where a Subject automatically notifies all subscribed Observers when its state changes. The Subject knows only the Observer interface — never the concrete types. Observers register dynamically, react independently, and can unsubscribe at any time. The result is a broadcast communication mechanism where adding new reactions to state changes requires no changes to the Subject or existing Observers.

🔑 Key Benefits:

• Open/Closed Principle: Add new observers without touching the subject or existing observers
• Loose coupling: Subject and observers depend only on interfaces, never on each other's implementations
• Dynamic subscription: Observers join and leave at runtime — flexible, responsive systems
• Broadcast communication: One state change automatically reaches any number of observers
• Separation of concerns: Subject manages state; each observer manages its own reaction

🌐 Language-Specific Highlights:

• Python: Always iterate a list() snapshot during notify() — safe if observers modify the list; use weakref.ref for observers that should be GC'd when unused; wrap each observer.update() in try/except to prevent one bad observer crashing all others
• TypeScript: Use Set<Observer> instead of Array to prevent duplicate subscriptions; schedule slow observer work asynchronously to avoid blocking the notification loop; type event objects as readonly interfaces for immutability guarantees
• Java: Use new ArrayList<>(observers) copy inside notifyObservers() to prevent ConcurrentModificationException; avoid the deprecated java.util.Observable — implement your own typed interfaces; use CopyOnWriteArrayList for concurrent notification without copying manually
• JavaScript: Store function references before subscribing if you need to unsubscribe later — anonymous functions can't be removed; store bound handler references explicitly to remove addEventListener correctly; use private class fields (#observers) to prevent external manipulation of the observer list
• C#: Always use the event keyword for delegate fields — prevents external invocation and clearing; implement IDisposable on observers that hold subscriptions and unsubscribe in Dispose(); use EventHandler<TEventArgs> for the standard .NET event pattern
• PHP: Use array_values() after array_filter() on the observer list to re-index sequential keys; always unsubscribe using the same object reference (PHP identity is reference-based); use SplObjectStorage for efficient object-keyed observer tracking
• Go: Protect the observer slice with sync.RWMutex — concurrent subscribe/notify causes data races; take a snapshot copy under a read lock before notifying, then notify without holding the lock; use non-blocking channel sends in channel-based observers to prevent hanging the notification loop
• Rust: Use Rc<RefCell<dyn Observer>> for shared ownership of observers; clone the data before the iteration loop to avoid borrow checker conflicts; for multi-threaded use, replace Rc<RefCell> with Arc<Mutex>
• Dart: Use StreamController.broadcast() for multiple-subscriber stream patterns; always iterate List.from(_observers) copy in notify(); wrap each observer call in try/catch to prevent one throwing observer stopping all others
• Swift: Use a WeakObserver wrapper struct to store weak references in the observer list — prevents retain cycles; prune nil weak references during notify(); use DispatchQueue or Swift actor to synchronize observer list access in concurrent code
• Kotlin: Always call .toList() on the observer list before iterating in notify() — prevents ConcurrentModificationException if an observer modifies the list; use Kotlin Flow or SharedFlow for the modern coroutine-compatible Observer; use ObservableProperty delegate only for simple single-listener cases

📋 When to Use Observer:

• One object's state change should automatically trigger reactions in any number of other objects
• You don't know at design time how many objects will need to react to state changes (or the number changes at runtime)
• You want components to react to events without being tightly coupled to the event source
• You need to broadcast the same event to multiple independent consumers
• You're implementing MVC/MVP where the model notifies views, event systems, reactive UIs, or pub/sub notification services

⚠️ When NOT to Use Observer:

• Only one observer is ever needed — a direct method call is simpler and clearer
• Order of notification matters and observers have dependencies on each other — use Mediator or a pipeline
• The caller needs a return value — Observer is fire-and-forget; use a direct call or Command instead
• The notification chain is hard to trace and debug is already a concern — explicit calls are easier to follow
• The system is small and simple — Observer infrastructure adds overhead where none is needed

🛠️ Best Practices Across Languages:

1. Notify from a snapshot copy of the observer list: Observers modifying the list during notification is a common bug — always iterate a copy
2. Wrap each observer notification in try/catch: One bad observer must never stop the others from being notified
3. Use event objects, not raw values: Typed event objects carry context (old value, new value, timestamp, event type) without coupling observers to the subject's full interface
4. Always clear subscriptions when done: Unsubscribe in lifecycle methods, Dispose(), deinit, or onDestroy() to prevent memory leaks
5. Use Set instead of List for the observer collection: Prevents accidental duplicate subscriptions and idempotent subscribe/unsubscribe
6. Never block the notification loop with slow work: Schedule heavy observer work asynchronously — the subject should not wait for observers to complete
7. Use weak references for long-lived subjects with short-lived observers: Prevents observers from being kept alive by the subject after their natural lifecycle ends
8. Keep observers focused: Each observer should do one thing and do it well — avoid fat observers that try to do everything

💡 Common Pitfalls:

• Concurrent modification: Modifying the observer list while iterating causes ConcurrentModificationException (Java), skipped observers (Python), or crashes (Go)
• Memory leaks from unremoved subscriptions: Subject holds a reference to a "dead" observer forever — prevents garbage collection
• Exception in one observer blocks the rest: Without try/catch per observer, one failure silences all downstream notifications
• Duplicate subscriptions: Subscribing the same observer twice causes it to fire twice — use Set to prevent this
• Infinite notification loops: Observer A's update() modifies Subject B, which notifies Observer C, which modifies Subject A → infinite loop. Guard with an _is_notifying flag or re-entrant protection.
• Stale data in the pull model: Observer pulls data from the subject after the notification; by that time, the subject may have changed again
• Retain cycles in Swift/ObjC: Subject holds strong observers; observers hold strong subjects — neither deallocates. Always use weak references on at least one side.
• Anonymous function unsubscription: Subscribing with an anonymous function/lambda creates a reference you can never match for unsubscription — always store the reference

🎁 Advanced Techniques:

• Reactive Streams (RxJava/RxJS/Kotlin Flow): Observer + composition operators + backpressure + error handling + scheduler control. The evolution of Observer for complex async data pipelines.
• SharedFlow / StateFlow (Kotlin): Kotlin's coroutine-native Observer implementations. StateFlow always replays the latest value to new subscribers (like a subject with state). SharedFlow is a hot broadcast stream with configurable replay.
• Combine (Apple): Swift's native reactive framework. Publisher is the subject; Subscriber is the observer. @Published property wrapper wraps a property in an AnyPublisher automatically.
• Debounce and throttle: Prevent rapid-fire notifications from overwhelming observers. A debounce waits N milliseconds of silence before forwarding; throttle forwards at most once per N milliseconds. Essential for search-as-you-type and window resize handlers.
• Priority-ordered notification: Sort observers by priority before notification — high-priority observers (audit loggers, circuit breakers) always run before low-priority ones.
• Event sourcing integration: Every domain event is an Observer notification. Persisting the event log (all notifications) gives you an audit trail and enables time-travel debugging.
• Subscription tokens / RAII subscriptions: Subscribe returns a disposable token. When the token goes out of scope, it automatically unsubscribes — no explicit lifecycle management needed. Used by RxJava's Disposable, Combine's AnyCancellable, and can be implemented in any language.

The Observer Pattern is the bedrock of reactive, event-driven software. Virtually every interactive application — from a weather widget to a trading system to a social feed — relies on it to connect state changes to the code that cares about them, without tying every part of the system to every other part.

Want to dive deeper into design patterns? Check out our comprehensive guides on:

• Mediator Pattern
• Iterator Pattern
• Command Pattern
• State Pattern
• Event-Driven Architecture Each guide includes examples in all 11 programming languages with language-specific best practices.