观察者模式和订阅模式

观察者模式和订阅模式在概念上是相似的，它们都涉及到一个对象（通常称为“主题”或“发布者”）和多个依赖对象（称为“观察者”或“订阅者”）之间的关系。然而，尽管它们有相似之处，但在某些方面也存在细微的差别。

观察者模式（Observer Pattern）

‌核心思想‌

定义对象间的一种一对多的依赖关系，使得每当一个对象改变状态，则所有依赖于它的对象都会得到通知并自动更新。

‌结构‌

通常包括主题（Subject）和观察者（Observer）两个主要角色。主题维护一个观察者列表，当状态变化时，通知列表中的所有观察者。

‌实现方式‌

观察者模式可以通过在主题中维护一个观察者列表，并提供注册（addObserver）和注销（removeObserver）观察者的方法来实现。当主题状态变化时，遍历观察者列表并调用每个观察者的更新方法。

‌应用场景‌

常用于事件处理系统、GUI工具包中的事件监听器、订阅-发布系统等。

Demo

设计一个天气预报系统，当天气变化时，通知多个订阅了天气预报的用户。

定义 Subject 接口

Subject接口，被观察者，代表被观察的对象，定义注册新观察者，移除观察者，和通知观察者三个接口。

package org.example.observer;import java.util.Observer;public interface MySubject {void registerObserver(Observer o);void removeObserver(Observer o);void notifyObservers();
}

实现 Subject 接口

储存观察者列表，被观察对象变化信息，通知观察者列表

package org.example.observer;import java.util.ArrayList;
import java.util.List;public class WeatherStation implements MySubject {private List<MyObserver> observers;private String weather;public WeatherStation() {observers = new ArrayList<MyObserver>();}@Overridepublic void registerObserver(MyObserver o) {this.observers.add(o);}@Overridepublic void removeObserver(MyObserver o) {this.observers.remove(o);}@Overridepublic void notifyObservers() {this.observers.forEach(observer -> observer.update(weather));}public void setWeather(String weather) {this.weather = weather;notifyObservers();}
}

定义 Observer 接口

Observer接口，代表观察者，定义接收到通知后需要执行的动作接口

package org.example.observer;public interface MyObserver {void update(String weather);
}

实现 Observer 接口

观察者接收到通知后，实现具体要执行的动作

package org.example.observer;public class User implements MyObserver {private String name;public User(String name) {this.name = name;}@Overridepublic void update(String weather) {System.out.println(String.format("name %s receive weather update : %s", name, weather));}
}

测试

package org.example.observer;public class MyObserverMain {public static void main(String[] args) {// 创建被观察者对象WeatherStation station = new WeatherStation();// 创建观察者对象User userA = new User("A");User userB = new User("B");User userC = new User("C");// 注册观察者，并更新天气station.registerObserver(userA);station.registerObserver(userB);station.registerObserver(userC);station.setWeather("Sunny");// 移除部分观察者，再次更新天气station.removeObserver(userA);station.setWeather("Rainy");}
}

订阅模式（Subscription Pattern）

‌核心思想‌

也是一种一对多的关系，但更强调“订阅”的概念。订阅者订阅某个主题或频道，以接收该主题或频道发布的更新或消息。

‌结构‌

通常包括发布者（Publisher）、订阅者（Subscriber）和消息（Message）三个主要角色。订阅者通过订阅某个发布者来接收其发布的消息。

‌实现方式‌

订阅模式可以通过事件总线（Event Bus）、消息队列（Message Queue）或专门的订阅系统来实现。订阅者可以订阅特定的主题或频道，并接收该主题或频道发布的消息。

‌应用场景‌

广泛用于消息传递系统、事件驱动架构、分布式系统中的服务间通信等。

Demo

定义消息类

package org.example.publish_subscriber;public class Message {private String content;public Message(String content) {this.content = content;}public String getContent() {return content;}
}

定义订阅者接口

package org.example.publish_subscriber;public interface Subscriber {void receive(String message);
}

实现订阅者接口

package org.example.publish_subscriber;public class User implements Subscriber{private String name;public User(String name) {this.name = name;}@Overridepublic void receive(String message) {System.out.println(String.format("%s received message: %s", name, message));}
}

定义事件总线

package org.example.publish_subscriber;import java.util.Map;
import java.util.concurrent.*;public class EventBus {private final Map<Subscriber, BlockingQueue<Message>> subscriberQueues = new ConcurrentHashMap<>();private final ExecutorService executor = Executors.newCachedThreadPool();private volatile boolean running = true;public void subscriber(Subscriber subscriber) {BlockingQueue<Message> queue = new LinkedBlockingQueue<>();subscriberQueues.put(subscriber, queue);executor.submit(() -> {try {while (running || !queue.isEmpty()) {Message message = queue.take();subscriber.receive(message);}} catch (Exception e) {Thread.currentThread().interrupt();}});}public void publish(Message message) {subscriberQueues.values().forEach(queue -> {try {queue.put(message);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}public void shutdown() {running = false;executor.shutdown();try {if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {executor.shutdownNow();}} catch (InterruptedException e) {executor.shutdownNow();Thread.currentThread().interrupt();}}
}

测试

package org.example.publish_subscriber;public class PubSubMain {public static void main(String[] args) throws InterruptedException {EventBus eventBus = new EventBus();User userA = new User("A");User userB = new User("B");User userC = new User("C");User userD = new User("D");eventBus.subscriber(userA);eventBus.subscriber(userB);eventBus.subscriber(userC);eventBus.subscriber(userD);for (int i = 0; i < 10; i++) {eventBus.publish(new Message(i + "Hello World!"));eventBus.publish(new Message(i + "Hello Shore!"));}Thread.sleep(10000);eventBus.shutdown();}
}