每天一个设计模式之观察者模式+发布订阅模式（Observer Pattern）

观察者模式定义了一套机制，可以实现一对多的关系，当“一”（被观察者）的状态发生了变化，会通知“多”个（观察者），这是行为型模式。

一、场景示例

这个模式存在这样的一个问题：如果由观察者不停地查询被观察者的状态变化，那么观察者将苦不堪言；而如果由被观察者来通知所有的观察者，那么有些对该状态不感兴趣的观察者将会“抱怨”。

二、观察者模式 vs 发布-订阅模式

1. 观察者模式，Subject和Observer是紧耦合的关系，而pub-sub是松散耦合，中间隔着event channel（或者称message broker）；
2. sub-obs直接发送消息，属于同步；而pub-sub则中间有一个消息队列，属于异步；
3. sub-obs缺点多多，Java 9已经去掉了官方观察者模式的实现；而pub-sub适用范围更广，在分布式系统中更是常见。

二、代码示例 - 观察者模式

观察者抽象类

public abstract class Observer {
    protected Subject subject;
    public abstract void update();
}
1
2
3
4

具体观察者

public class HexObserver extends Observer {

    public HexObserver(Subject subject) {
        this.subject = subject;
        this.subject.attach(this);
    }

    @Override
    public void update() {
        System.out.println("Hex string: " + subject.getState());
    }
}

public class OctalObserver extends Observer {

    public OctalObserver(Subject subject) {
        this.subject = subject;
        this.subject.attach(this);
    }

    @Override
    public void update() {
        System.out.println("Octal string: " + subject.getState());
    }
}

public class BinaryObserver extends Observer {

    public BinaryObserver(Subject subject) {
        this.subject = subject;
        this.subject.attach(this);
    }

    @Override
    public void update() {
        System.out.println("Binary string: " + subject.getState());
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

被观察者（主题类）

public class Subject {
    private int state;
    private List<Observer> observers = new ArrayList<>();

    public int getState() {
        return state;
    }

    public void setState(int state) {
        this.state = state;
        notifyAllObservers();
    }

    public void attach(Observer observer) {
        observers.add(observer);
    }

    public void notifyAllObservers() {
        for (Observer o : observers) {
            o.update();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

客户类

public class Client {
    public static void main(String[] args) {
        Subject subject = new Subject();

        new HexObserver(subject);
        new BinaryObserver(subject);
        new OctalObserver(subject);

        System.out.println("First state changes: 111");
        subject.setState(111);

        System.out.println("--------------------------");

        System.out.println("Second state changes: 222");
        subject.setState(222);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

运行结果

First state changes: 111
Hex string: 111
Binary string: 111
Octal string: 111
--------------------------
Second state changes: 222
Hex string: 222
Binary string: 222
Octal string: 222

Process finished with exit code 0

1
2
3
4
5
6
7
8
9
10
11
12

三、代码示例 - 发布-订阅模式

发布者

public class Publisher {

    public Publisher(Integer pubId) {
        this.pubId = pubId;
    }

    private Integer pubId;

    public Integer getPubId() {
        return pubId;
    }

    public void setPubId(Integer pubId) {
        this.pubId = pubId;
    }

    @Override
    public String toString() {
        return "Publisher{" +
                "pubId=" + pubId +
                '}';
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

订阅者

public class Subscriber {
    public Subscriber(Integer subId) {
        this.subId = subId;
    }

    private Integer subId;

    public Integer getSubId() {
        return subId;
    }

    public void setSubId(Integer subId) {
        this.subId = subId;
    }

    @Override
    public String toString() {
        return "Subscriber{" +
                "subId=" + subId +
                '}';
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

消息类

public class Message {
    public Message(String content) {
        this.content = content;
    }

    private String content;

    @Override
    public String toString() {
        return "Message{" +
                "content='" + content + '\'' +
                '}';
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

发布-订阅控制中心

public interface SubPubCentral {
    boolean subscribe(Publisher publisher, Subscriber subscriber);
    boolean unsubscribe(Publisher publisher, Subscriber subscriber);
    void publish(Publisher publisher, Message message);
}

public class SubPubCentralImpl implements SubPubCentral {
    private static Map<Integer, Set<Integer>> pubSubMap;

    static {
        pubSubMap = new HashMap<>();
    }
    @Override
    public boolean subscribe(Publisher publisher, Subscriber subscriber) {
        Set<Integer> subscribeSet = pubSubMap.getOrDefault(publisher.getPubId(), new HashSet<>());
        boolean add = subscribeSet.add(subscriber.getSubId());
        if (add) pubSubMap.put(publisher.getPubId(), subscribeSet);
        return pubSubMap.size() > 0;
    }

    @Override
    public boolean unsubscribe(Publisher publisher, Subscriber subscriber) {
        Set<Integer> subscribeSet = pubSubMap.get((publisher.getPubId()));
        boolean remove = false;
        if (subscribeSet != null && subscribeSet.size() > 0) {
            remove = subscribeSet.remove(subscriber.getSubId());
            if (remove) pubSubMap.put(publisher.getPubId(), subscribeSet);
        }
        return remove;
    }

    @Override
    public void publish(Publisher publisher, Message message) {
        Set<Integer> subscribeSet = pubSubMap.get((publisher.getPubId()));
        for (Integer id : subscribeSet) {
            System.out.println("向发布者[" + publisher.getPubId() + "]的订阅者[" + id + "]发送消息：" + message);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

发布控制器和订阅控制器

public class PublisherController {
    private SubPubCentral subPubCentral;

    public PublisherController(SubPubCentral subPubCentral) {
        this.subPubCentral = subPubCentral;
    }

    public void publish(Integer pubId, String message) {
        subPubCentral.publish(new Publisher(pubId), new Message(message));
    }
}

public class SubscribeController {
    private SubPubCentral subPubCentral;

    public SubscribeController(SubPubCentral subPubCentral) {
        this.subPubCentral = subPubCentral;
    }

    public void sub(Integer subId, Integer pubId) {
        subPubCentral.subscribe(new Publisher(pubId), new Subscriber(subId));
    }

    public void unsub(Integer subId, Integer pubId) {
        subPubCentral.unsubscribe(new Publisher(pubId), new Subscriber(subId));
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

客户类

public class Client {
    public static void main(String[] args) {
        SubPubCentral subPubCentral = new SubPubCentralImpl();

        PublisherController publisherController = new PublisherController(subPubCentral);
        SubscribeController subscribeController = new SubscribeController(subPubCentral);

        subscribeController.sub(100, 001);
        subscribeController.sub(101, 001);

        publisherController.publish(001, "from#001 publish, COVID-19 regions of high risk");
        System.out.println("~~~~~~~~~~~~~~~~~~");

        subscribeController.unsub(100, 001);
        publisherController.publish(100, "$$%%$$");
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

【参考】

1. https://blog.csdn.net/David_TD/article/details/107715396
2. https://embeddedartistry.com/fieldatlas/differentiating-observer-and-publish-subscribe-patterns/
3. https://www.runoob.com/design-pattern/observer-pattern.html
4. https://refactoring.guru/design-patterns/observer