当前位置：首页 > article >正文

观察者模式和订阅模式

article 2024/11/23 19:10:32

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

观察者模式（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();
    }
}