观察者模式和订阅模式在概念上是相似的，它们都涉及到一个对象（通常称为“主题”或“发布者”）和多个依赖对象（称为“观察者”或“订阅者”）之间的关系。然而，尽管它们有相似之处，但在某些方面也存在细微的差别。
观察者模式（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>();
}
@Override
public void registerObserver(MyObserver o) {
this.observers.add(o);
}
@Override
public void removeObserver(MyObserver o) {
this.observers.remove(o);
}
@Override
public 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;
}
@Override
public 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;
}
@Override
public 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();
}
}
在上面的例子中，executor.submit 被用于提交订阅者线程的任务。每个订阅者都有一个对应的工作队列，当发布消息时，消息会被放入每个订阅者的工作队列中。订阅者线程会不断地从自己的工作队列中取出消息并处理。通过这种方式，实现了发布-订阅模式中的异步通信和消息分发
区别与联系
区别
观察者模式更侧重于对象间的依赖关系和状态变化的通知机制；而订阅模式更强调消息的传递和订阅-发布的关系。此外，观察者模式通常是在单个应用或系统内使用；而订阅模式可能涉及跨系统或跨网络的消息传递。
联系
两者都涉及到一个对象（主题/发布者）和多个依赖对象（观察者/订阅者）之间的关系，且都实现了某种形式的通知机制。在某些情况下，它们可以相互替代或结合使用。例如，在一个分布式系统中，可以使用订阅模式来实现不同服务之间的消息传递，而在服务内部则可以使用观察者模式来实现状态变化的通知和更新。