Observer Design Pattern - Publishing data to multiple subscribers

Published on August 11, 2026

Subscribe to the Newsletter

Join other readers for the latest posts and insights on AI, MLOps, and best practices in software design.

The Observer Design Pattern is useful when one component produces information that several other components need to process independently.

Consider a machine telemetry application. A machine continuously produces measurements such as temperature, vibration, and pressure. Several parts of the application are interested in these measurements, but each uses them differently.

An EventLogger writes every measurement to a log. An AverageReport collects several measurements and calculates average values. A SafetyMonitor checks the measurements against configured limits and reports potentially unsafe conditions.

The MachineTelemetryReporter is the publisher of these measurements. It obtains new MachineMeasurementEvent objects from a MachineDataSource and makes them available to the other components.

Publisher-Subscriber Model — Figure 1: A simple Publisher-Subscriber model where the publisher sends updates to multiple subscribers.

Question: Are the reports just different ways of processing the same measurements?

Answer: Yes, the reports are different ways of processing the same measurements. Each component, such as the EventLogger, AverageReport, and SafetyMonitor, uses the measurements in a unique way to achieve its specific purpose.

Before applying the Observer Design Pattern, let us first implement this application in the most direct way.

Before applying the Observer Design Pattern

The initial version of the application does not use a common abstraction for the consumers. Instead, the MachineTelemetryReporter directly knows the three classes that process its measurements.

The class diagram in the following figure shows this first design, in which the MachineDataSource produces a new MachineMeasurementEvent whenever its read() method is called. The MachineTelemetryReporter repeatedly requests these events and passes each event to the three consumer objects. There is one detail in the diagram that is easy to overlook. Each consumer exposes a different method:

EventLogger.log_event() -> logs the event
AverageReport.add_measurement() -> adds the measurement
SafetyMonitor.inspect() -> inspects the measurement

This can easily happen in a real project. The classes may have been written by different developers, and the developers may simply have chosen names that made sense for their individual components.

Initial architecture before applying Observer Design Pattern — Figure 2: Initial architecture before applying the Observer Design Pattern, showing direct dependencies between the publisher and its consumers.

The consequence is that MachineTelemetryReporter must know how every consumer is implemented.

Each machine measurement is represented by a MachineMeasurementEvent.

@dataclass(frozen=True)
class MachineMeasurementEvent:
    machine_id: str
    temperature: float
    vibration: float
    pressure: float
    timestamp: datetime

The event contains the values belonging to one measurement and identifies the machine from which the data originated.

To produce test data, we need a data source that generates MachineMeasurementEvent objects. For this example, MachineDataSource generates random values. This gives us a simple way to test the application without connecting it to an actual machine.

class MachineDataSource:
 
    def __init__(self, machine_id: str) -> None:
        self._machine_id = machine_id
 
    def read(self) -> MachineMeasurementEvent:
        return MachineMeasurementEvent(
            machine_id=self._machine_id,
            temperature=random.uniform(20.0, 100.0),
            vibration=random.uniform(-0.050, 0.050),
            pressure=random.uniform(1.0, 5.0),
            timestamp=datetime.now(),
        )

Every call to read() creates and returns a new measurement event.

The three consumers process the same event in different ways. Starting with the simplest, the EventLogger processes each measurement immediately.

class EventLogger:
 
    def log_event(
        self,
        event: MachineMeasurementEvent,
    ) -> None:
        print(
            f"[LOG] {event.machine_id}: "
            f"T={event.temperature:.1f} °C, "
            f"V={event.vibration:.3f}, "
            f"P={event.pressure:.2f} bar"
        )

Parallel to the event logger, the AverageReport processes the same event in a different way. Instead of printing it immediately, it stores measurements so that statistics can later be calculated.

class AverageReport:
 
    def __init__(self) -> None:
        self._events: list[MachineMeasurementEvent] = []
        self._report_every = 5
 
    def add_measurement(
        self,
        event: MachineMeasurementEvent,
    ) -> None:
        self._events.append(event)
 
        if len(self._events) >= self._report_every:
            self._print_report()
            self._events.clear()
 
    def _print_report(self) -> None:
        count = len(self._events)
 
        avg_temperature = sum(event.temperature for event in self._events) / count
 
        avg_vibration = sum(event.vibration for event in self._events) / count
 
        avg_pressure = sum(event.pressure for event in self._events) / count
 
        print(
            f"[REPORT] Average of {count} events | "
            f"T={avg_temperature:.1f} °C | "
            f"V={avg_vibration:.2f} mm/s | "
            f"P={avg_pressure:.2f} bar"
        )

The last consumer is the SafetyMonitor. It also processes the same event, but it does not print every measurement. Instead, it checks whether the values exceed configured limits.

class SafetyMonitor:
 
    def __init__(
        self,
        max_temperature: float,
        max_vibration: float,
        max_pressure: float,
    ) -> None:
        self._max_temperature = max_temperature
        self._max_vibration = max_vibration
        self._max_pressure = max_pressure
 
    def inspect(self, event: MachineMeasurementEvent) -> None:
        problems: list[str] = []
 
        if event.temperature > self._max_temperature:
            problems.append("temperature too high")
 
        if event.vibration > self._max_vibration:
            problems.append("vibration too high")
 
        if event.pressure > self._max_pressure:
            problems.append("pressure too high")
 
        if problems:
            print(f"[ALARM] {event.machine_id}: " + ", ".join(problems))

The telemetry reporter aggregates the data source and the three consumers. It repeatedly reads new measurements and passes them to the consumers.

In this way the MachineTelemetryReporter brings the components together.

class MachineTelemetryReporter:
 
    def __init__(
        self,
        data_source: MachineDataSource,
        event_logger: EventLogger,
        average_report: AverageReport,
        safety_monitor: SafetyMonitor,
    ) -> None:
        self._data_source = data_source
        self._event_logger = event_logger
        self._average_report = average_report
        self._safety_monitor = safety_monitor
 
    def report_measurements(
        self,
        samples: int,
    ) -> None:
        for _ in range(samples):
            event = self._data_source.read()
 
            self._event_logger.log_event(event)
            self._average_report.add_measurement(event)
            self._safety_monitor.inspect(event)

The method obtains one measurement at a time and explicitly passes it to every consumer. And indeed the application works correctly. The MachineTelemetryReporter produces measurements, and the three consumers process them in their own way. The architecture, however, has several problems.

The consumers are hardcoded

MachineTelemetryReporter directly contains references to EventLogger, AverageReport, and SafetyMonitor.

Suppose we later want to add a DatabaseWriter. The constructor of MachineTelemetryReporter must change. Its attributes must change and report_measurements() must change:

self._database_writer.store(event)

Also removing a consumer requires another modification.

The publisher therefore cannot gain or lose subscribers without changing its own implementation.

The publisher knows how every consumer works

There is a second problem, which is that MachineTelemetryReporter must know that it should call log_event(event) for event logging, add_measurement(event) for average reporting, and inspect(event) for safety monitoring.

The reporter is therefore coupled to the concrete API of every consumer. The classes are not loosely coupled.

What we would prefer is a design in which the reporter knows only that there are components interested in new measurements. It should not need to know what those components are or how they process the measurements.

After applying Observer

The Observer Design Pattern addresses this problem by introducing two roles. The Subject represents the publisher and the Observer represents a subscriber that is interested in changes produced by the Subject. In our redesigned application, MachineTelemetryReporter becomes a subclass of Subject.

EventLogger, AverageReport, and SafetyMonitor all implement the Observer interface.

Architecture after applying Observer Design Pattern — Figure 3: Architecture after applying the Observer Design Pattern, showing the Subject-Observer relationship.

The most important difference from the previous diagram is that MachineTelemetryReporter no longer has direct references to the three concrete consumers. Instead, Subject maintains a collection of Observer objects. At runtime, observers can be added to or removed from that collection.

Every Observer also exposes the same operation update() by which the Subject can notify it of new measurements. The publisher therefore no longer needs to know whether the concrete object is a logger, report generator, safety monitor, or something that may be introduced in the future.

The Observer interface defines the method that every subscriber must implement.

from abc import ABC, abstractmethod
 
 
class Observer(ABC):
 
    @abstractmethod
    def update(
        self,
        event: MachineMeasurementEvent,
    ) -> None:
        pass

The concrete observers can process the event in any way they want, but the Subject can notify all of them through the same method.

The Subject superclass Subject keeps track of the currently subscribed observers.

class Subject:
 
    def __init__(self) -> None:
        self._observers: list[Observer] = []
 
    def attach(self, observer: Observer) -> None:
        if observer not in self._observers:
            self._observers.append(observer)
 
    def detach(self, observer: Observer) -> None:
        if observer in self._observers:
            self._observers.remove(observer)
 
    def _notify(
        self,
        event: MachineMeasurementEvent,
    ) -> None:
        for observer in self._observers:
            observer.update(event)

An Observer subscribes through attach() and can unsubscribe through detach().

Whenever the Subject has new content, _notify() iterates over the subscribed observers and calls their public update() method.

The important difference from the first implementation is that this method no longer contains knowledge about concrete consumer classes.

It simply calls:

observer.update(event)

The concrete publisher MachineTelemetryReporter inherits the subscription behavior from Subject.

class MachineTelemetryReporter(Subject):
 
    def __init__(
        self,
        data_source: MachineDataSource,
    ) -> None:
        super().__init__()
        self._data_source = data_source
 
    def report_measurements(
        self,
        samples: int,
    ) -> None:
        for _ in range(samples):
            event = self._data_source.read()
            self._notify(event)

The reporter now has a much narrower responsibility. It obtains measurements and publishes them. It does not know what happens to those measurements afterward.

The three consumers are now implemented as Observers. As shown in the code below, EventLogger implements Observer, so it provides the update() method.

class EventLogger(Observer):
 
    def __init__(
        self,
        subject: Subject,
    ) -> None:
        self._subject = subject
        self._subject.attach(self)
 
    def update(
        self,
        event: MachineMeasurementEvent,
    ) -> None:
        print(
            f"[LOG] {event.timestamp:%H:%M:%S} | "
            f"{event.machine_id} | "
            f"T={event.temperature:.1f} °C | "
            f"V={event.vibration:.3f} | "
            f"P={event.pressure:.2f} bar"
        )
 
    def close(self) -> None:
        self._subject.detach(self)

The constructor receives a reference to the Subject and registers the logger as an observer. The logger can later remove itself by calling detach().

The AverageReport class is also an Observer. It implements the same update() method, but its processing logic is entirely different.

class AverageReport(Observer):
 
    def __init__(
        self,
        subject: Subject,
        report_every: int = 5,
    ) -> None:
        self._subject = subject
        self._subject.attach(self)
 
        self._report_every = report_every
        self._events: list[MachineMeasurementEvent] = []
 
    def update(
        self,
        event: MachineMeasurementEvent,
    ) -> None:
        self._events.append(event)
 
        if len(self._events) >= self._report_every:
            self._print_report()
            self._events.clear()
 
    def _print_report(self) -> None:
        count = len(self._events)
 
        avg_temperature = sum(
            event.temperature for event in self._events
        ) / count
 
        avg_vibration = sum(
            event.vibration for event in self._events
        ) / count
 
        avg_pressure = sum(
            event.pressure for event in self._events
        ) / count
 
        print(
            f"[REPORT] Average of {count} measurements | "
            f"T={avg_temperature:.1f} °C | "
            f"V={avg_vibration:.3f} | "
            f"P={avg_pressure:.2f} bar"
        )
 
    def close(self) -> None:
        self._subject.detach(self)

The reporter does not know that this Observer collects several measurements before producing output. That decision belongs entirely to AverageReport.

The last consumer, SafetyMonitor, also implements the same Observer interface.

class SafetyMonitor(Observer):
 
    def __init__(
        self,
        subject: Subject,
        max_temperature: float = 95.0,
        max_vibration: float = 0.045,
        max_pressure: float = 4.5,
    ) -> None:
        self._subject = subject
        self._subject.attach(self)
 
        self._max_temperature = max_temperature
        self._max_vibration = max_vibration
        self._max_pressure = max_pressure
 
    def update(self, event: MachineMeasurementEvent) -> None:
        problems: list[str] = []
 
        if event.temperature > self._max_temperature:
            problems.append("temperature too high")
 
        if event.vibration > self._max_vibration:
            problems.append("vibration too high")
 
        if event.pressure > self._max_pressure:
            problems.append("pressure too high")
 
        if problems:
            print(f"[ALARM] {event.machine_id}: " + ", ".join(problems))
 
    def close(self) -> None:
        self._subject.detach(self)

The three subscribers now have the same public notification method:

EventLogger.update()
AverageReport.update()
SafetyMonitor.update()

but they remain responsible for processing the measurement in their own way.

The test program creates the publisher and passes it to the Observer constructors and runs the measurement reporting loop.

def main() -> None:
    data_source = MachineDataSource(machine_id="machine-001")
 
    reporter = MachineTelemetryReporter(data_source)
 
    event_logger = EventLogger(subject=reporter)
    average_report = AverageReport(subject=reporter)
    safety_monitor = SafetyMonitor(
        subject=reporter,
        max_temperature=95.0,
        max_vibration=0.045,
        max_pressure=4.5,
    )
 
    reporter.report_measurements(samples=10)
 
    event_logger.close()
    average_report.close()
    safety_monitor.close()
 
 
if __name__ == "__main__":
    main()

Each Observer registers itself with the publisher during construction. When report_measurements() starts producing events, the inherited _notify() method distributes each event to the currently subscribed observers.

What did the Observer Pattern change?

The first implementation contained three explicit calls:

self._event_logger.log_event(event)
self._average_report.add_measurement(event)
self._safety_monitor.inspect(event)

The Observer-based implementation replaces them with:

self._notify(event)

This small change reflects a much larger architectural difference - MachineTelemetryReporter no longer hardcodes its subscribers.

New Observer objects can be attached at runtime without modifying the publisher.

Existing observers can detach when they are no longer interested.

Most importantly, the publisher knows nothing about how an Observer processes a measurement. It knows only that every Observer supports:

update(event)

The logger may print the measurement. The report may collect it. The safety monitor may ignore most measurements and react only when a limit is exceeded.

Those decisions belong to the observers.

The publisher's job is simply to produce the measurement and notify its subscribers.

And so my friends, this is the separation of concerns that the Observer Design Pattern provides. The publisher is concerned only with producing measurements, and the subscribers are concerned only with processing them.

Conclusion

I hope you enjoyed this example of the Observer Design Pattern. It is a simple example, but it illustrates the key concepts and benefits of using this pattern in software design.

If you have any questions or want to see more examples like that, feel free to reach out or check out the source code here.

If you came across this pattern in your work, I would love to hear about your experiences and how you applied it in your projects.

Thanks for reading!

– Richard