【設計模式（19）】行為型模式之觀察者模式

阿新 • • 發佈：2021-06-29

個人學習筆記分享，當前能力有限，請勿貶低，菜鳥互學，大佬繞道

如有勘誤，歡迎指出和討論，本文後期也會進行修正和補充

前言

前面有一篇文章已經介紹了訂閱/釋出模式，即生產者和消費者通過一箇中介者來互動

生產者只負責向中介傳遞資料，不關心其餘步驟
消費者在中介者處進行註冊，告知中介者自己需要資料
中介者接受來自生產者的資料，並傳遞給在自己這裡註冊過的消費者

當生產者只有一個的時候，可以省略掉中介者，直接在生產者處註冊消費者

通常滿足N-1-N或者1-N的互動模型

消費者在中介者處或者直接向生產者訂閱訊息，而生產者負責釋出訊息，由中介者或者生產者

因而被稱為訂閱/釋出模式

可以看到，註冊過的消費者總是在等待訊息，無論訊息來自中介者，或者直接來源於生產者，最終目的都是觀察生產者

因此這種模式也被稱為觀察者模式

在實際生活中，最常見的就是訂閱，無論是簡訊訂閱，還是微信上的訂閱號，我們都是在作為消費者，被動的接受訊息（雖然很多時候都是單方面在騷擾我們。。。）

而在開發中，生產者負責生產訊息，不關心如何被消費以及消費者是誰；消費者註冊並接受訊息，不關心訊息的來源和時間；生產者和消費者並不需要時刻保持聯絡

其核心目的還是那個老生常談的，解耦

1.介紹

適用目的：定義物件間的一種一對多的依賴關係，當一個物件的狀態發生改變時，所有依賴於它的物件都得到通知並被自動更新。

主要解決：一個物件狀態改變給其他物件通知的問題，而且要考慮到易用和低耦合，保證高度的協作。

何時使用：一個物件（目標物件）的狀態發生改變，所有的依賴物件（觀察者物件）都將得到通知，進行廣播通知。

如何解決：使用面向物件技術，可以將這種依賴關係弱化。

關鍵程式碼：在抽象類裡有一個集合存放觀察者們。

應用例項：簡訊/公眾號推送；平臺的公告；股票與股民；

優點：

降低了目標與觀察者之間的耦合關係，兩者之間是抽象耦合關係
目標與觀察者之間建立了一套觸發機制。

缺點：

目標與觀察者之間的依賴關係並沒有完全解除，而且有可能出現迴圈引用。
當觀察者物件很多時，通知的釋出會花費很多時間，影響程式的效率

使用場景：

一個抽象模型有兩個方面，其中一個方面依賴於另一個方面。將這些方面封裝在獨立的物件中使它們可以各自獨立地改變和複用。
一個物件的改變將導致其他一個或多個物件也發生改變，而不知道具體有多少物件將發生改變，可以降低物件之間的耦合度。

一個物件必須通知其他物件，而並不知道這些物件是誰。
需要在系統中建立一個觸發鏈，A物件的行為將影響B物件，B物件的行為將影響C物件……，可以使用觀察者模式建立一種鏈式觸發機制。

注意事項：

避免迴圈引用
非同步以防止某一個觀察者出錯導致整個系統卡殼

2.結構

觀察者模式的主要角色

抽象主題（Subject）：也叫抽象目標類，它提供了一個用於儲存觀察者物件的聚集類和增加、刪除觀察者物件的方法，以及通知所有觀察者的抽象方法。
具體主題（Concrete Subject）：也叫具體目標類，它實現抽象目標中的通知方法，當具體主題的內部狀態發生改變時，通知所有註冊過的觀察者物件。
抽象觀察者（Observer）：它是一個抽象類或介面，它包含了一個更新自己的抽象方法，當接到具體主題的更改通知時被呼叫。
具體觀察者（Concrete Observer） ：實現抽象觀察者中定義的抽象方法，以便在得到目標的更改通知時更新自身的狀態。

3.步驟

建立抽象目標

// 抽象目標
abstract class Subject {
    protected Collection<Observer> observers = new HashSet<>();

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

    public void remove(Observer observer) {
        observers.remove(observer);
    }

    public abstract void notifyObserver(String msg);
}

建立具體目標，繼承抽象目標，並實現其虛擬方法

// 具體目標
class ConcreteSubject extends Subject {
    @Override
    public void notifyObserver(String msg) {
        System.out.println("具體目標發生改變！" + msg);

        observers.parallelStream().forEach(m -> m.response(msg));
    }
}

建立抽象觀察者

// 抽象觀察者
interface Observer {
    void response(String msg);
}

建立具體觀察者，實現抽象觀察者介面

// 具體觀察者A
class ConcreteObserverA implements Observer {
    @Override
    public void response(String msg) {
        System.out.println("具體觀察者A作出反應！" + msg);
    }
}

// 具體觀察者B
class ConcreteObserverB implements Observer {
    @Override
    public void response(String msg) {
        System.out.println("具體觀察者B作出反應！" + msg);
    }
}

測試程式碼

public class ObserverTest {
    public static void main(String[] args) {
        Subject subject = new ConcreteSubject();
        Observer observerA = new ConcreteObserverA();
        Observer observerB = new ConcreteObserverB();
        subject.add(observerA);
        subject.add(observerB);
        subject.notifyObserver("hello world");
        subject.remove(observerA);
        subject.notifyObserver("你好");
    }
}

執行結果

4.擴充套件

實際上在Java的jdk中，已經通過 java.util.Observable 類和 java.util.Observer 介面定義了觀察者模式，只要實現他們的子類即可編寫觀察者模式例項

但是兩個已被jdk9棄用，官方推薦的做法是使用java.util.concurrent.Flow的API

下面會對這兩種分別給出示例

4.1.Observable類 + Observer類

Observable類是抽象目標類，持有一個Vector向量，用於儲存所有要通知的觀察者物件。

主要方法如下：

void addObserver(Observer o)：用於將新的觀察者物件新增到向量中
void notifyObservers(Object arg) ：呼叫向量中所有觀察者的update()方法，通知他們資料已發生改變。通常先通知後放入的觀察者；可以通過引數arg向update()傳遞資料
void setChange()：用於設定一個布林型別的內部標誌位，註明目標物件已發生改變；當它為真時，notifyObservers才會通知觀察者

完整示例如下

package com.company.designPattern.observer;

import java.util.Date;
import java.util.Observable;
import java.util.Observer;

// 被觀察者（具體目標）
class NumObservable extends Observable {
    private int num = 0;

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
        setChanged();
        notifyObservers(new Date());
    }
}

// 觀察者A
class ObserverA implements Observer {
    @Override
    public void update(Observable o, Object arg) {
        NumObservable object = (NumObservable) o;
        System.out.println("ObserverA: Num has changed to " + object.getNum() + "\n Message: " + arg);
    }
}

// 觀察者B
class ObserverB implements Observer {
    @Override
    public void update(Observable o, Object arg) {
        NumObservable object = (NumObservable) o;
        System.out.println("ObserverB: Num has changed to " + object.getNum() + "\n Message: " + arg);
    }
}

public class ObserverTest1 {
    public static void main(String[] args) {
        // 建立被觀察者和觀察者
        NumObservable observable = new NumObservable();
        Observer observerA = new ObserverA();
        Observer observerB = new ObserverB();
        // 關聯
        observable.addObserver(observerA);
        observable.addObserver(observerB);
        // 修改資料10
        observable.setNum(10);
        // 修改資料20
        observable.setNum(20);
        // 解除observerA的觀察關聯，修改資料30
        observable.deleteObserver(observerA);
        observable.setNum(30);
    }
}

執行結果

前兩次，按照後加入先通知的順序，分別通知了A和B

第三次，解除了A的關聯，所以只通知了B

4.2.Flow API

Flow API 是 Java 9 引入的響應式程式設計的介面，其中包含4個介面：

Publisher：釋出者，負責釋出訊息；
Subscriber：訂閱者，負責訂閱處理訊息；
Subscription：訂閱控制類，可用於釋出者和訂閱者之間通訊；
Processor：處理者，同時充當Publisher和Subscriber的角色。

請注意Flow API僅提供介面，並不提供具體實現，請自行按照需求實現

示例如下

定義一個類，用於訂閱者和釋出者之間傳輸資料

/**
 * 定義一個用於傳遞資料的類
 */
class Message {
    public String msg = "";
    public int leftCount = 0;

    public Message(String msg, int leftCount) {
        this.msg = msg;
        this.leftCount = leftCount;
    }
}

可以根據自己的需求構造類的內容

定義一個釋出者

/**
 * 自定義釋出者
 * 需要指定訂閱者傳送給釋出者的資料型別
 */
class MyPublisher implements Flow.Publisher<Message> {
    private int count = 0;          // 計數器，從0開始
    private final int maxCount;     // 最大計數器
    private int leftCount = 0;      // 剩餘計數
    private final long interval;    // 傳送間隔
    private boolean isCanceled;     // 是否被取消

    /**
     * 建構函式，根據需要初始化資料
     *
     * @param interval 初始化傳送間隔
     * @param maxCount 最大計數器，達到數量後自動停止
     */
    public MyPublisher(long interval, int maxCount) {
        this.interval = interval;
        this.maxCount = maxCount;
    }

    /**
     * 訂閱事件
     * 在這裡定義訂閱者訂閱後的操作，通常是在某條件下傳遞一個物件給訂閱者
     * 為方便演示，我們每隔一段時間向訂閱者傳送當前計數N次，N由訂閱者傳遞給我們
     *
     * @param subscriber
     */
    @Override
    public void subscribe(Flow.Subscriber<? super Message> subscriber) {
        // 使用執行緒來非同步執行每個訂閱操作
        new Thread(() -> {
            try {
                // 給訂閱者分配一個控制器
                subscriber.onSubscribe(new MySubscription());
                // 迴圈執行核心操作
                while (!isCanceled && count < maxCount) {
                    // 當剩餘數量大於0時，傳遞資料給訂閱者
                    if (leftCount > 0) {
                        subscriber.onNext(new Message(new Date() + ":" + ++count, --leftCount));
                        Thread.sleep(interval);
                    }
                }
                // 結束訂閱後，通知訂閱者已結束
                subscriber.onComplete();
            } catch (Exception e) {
                // 出現錯誤時，通知訂閱者發生錯誤
                subscriber.onError(e);
            }
        }).start();
    }

    /**
     * 自定義訂閱控制類
     * 重寫request和cancel方法，提供給訂閱者使用
     */
    private class MySubscription implements Flow.Subscription {

        /**
         * 接受到來自訂閱者的資料請求
         *
         * @param n 請求次數
         */
        @Override
        public void request(long n) {
            // 將次數累加到剩餘次數中
            leftCount += n;
        }

        /**
         * 接收到來自訂閱者的取消請求
         */
        @Override
        public void cancel() {
            isCanceled = true;
        }
    }
}

釋出者的核心任務即subscribe，需要在這裡定義訂閱後的操作，通常非同步執行

定義一個訂閱者

/**
 * 自定義訂閱者
 * 需要指定從釋出者接收到的資料型別
 * 模擬事件：請求一定數量的資料，並且根據需要分批請求
 */
class MySubscriber implements Flow.Subscriber<Message> {
    private Flow.Subscription subscription; // 用於持有來自訂閱者的控制器（其實並不必要）
    private int perNum; // 每輪數量
    private int count;  // 計數器

    /**
     * 建構函式，根據需要初始化資料
     *
     * @param perNum 每輪訂閱次數
     * @param count  訂閱次數
     */
    public MySubscriber(int perNum, int count) {
        this.perNum = perNum;
        this.count = count;
    }

    /**
     * 發起一輪請求
     */
    private void startNewRound() {
        System.out.println("Start a new round");
        int requestCount = Math.min(count, perNum);
        count -= requestCount;
        subscription.request(requestCount);
    }

    /**
     * 訂閱事件
     *
     * @param subscription
     */
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        // 發起第一輪請求
        startNewRound();
    }

    // 接受來自發布者的觸發指令
    @Override
    public void onNext(Message item) {
        System.out.println("receive message: " + item.msg);
        System.out.println("now left: " + item.leftCount);
        // 本輪結束的時候，開啟下一輪
        if (item.leftCount == 0 && count > 0) {
            startNewRound();
        }
    }

    // 接受來自發布者的錯誤
    @Override
    public void onError(Throwable throwable) {
        System.out.println("onError:" + throwable.getMessage());
    }

    // 接受來自發布者的完成指令
    @Override
    public void onComplete() {
        System.out.println("onComplete!");
    }
}

核心部分為onSubscribe和onNext，分別用於發起第一次請求，和發起後續請求

客戶端程式碼

public class FlowDemo {
    public static void main(String[] args) {
        MyPublisher publisher = new MyPublisher(500L, 10);      // 每500ms傳送一次，最多20次
        MySubscriber subscriber = new MySubscriber(3, 20);        //每輪發送3次，總共8輪
        publisher.subscribe(subscriber);
        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

執行結果