異步非阻塞框架是如何實(shí)現(xiàn)的

1、什么是觀察者模式？

Define a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically.

觀察者模式（Observer Design Pattern）：在對象之間定義一個(gè)一對多的依賴，當(dāng)一個(gè)對象狀態(tài)改變的時(shí)候，所有依賴的對象都會得到通知并自動更新。

說人話：也叫發(fā)布訂閱模式，能夠很好的解耦一個(gè)對象改變，自動改變另一個(gè)對象這種情況。

2、觀察者模式定義

①、Subject 被觀察者

定義被觀察者必須實(shí)現(xiàn)的職責(zé)， 它必須能夠動態(tài)地增加、 取消觀察者。它一般是抽象類或者是實(shí)現(xiàn)類， 僅僅完成作為被觀察者必須實(shí)現(xiàn)的職責(zé)：管理觀察者并通知觀察者。

②、Observer觀察者

觀察者接收到消息后， 即進(jìn)行update（更新方法） 操作， 對接收到的信息進(jìn)行處理。

③、ConcreteSubject具體的被觀察者

定義被觀察者自己的業(yè)務(wù)邏輯， 同時(shí)定義對哪些事件進(jìn)行通知。

④、ConcreteObserver具體的觀察者

每個(gè)觀察在接收到消息后的處理反應(yīng)是不同， 各個(gè)觀察者有自己的處理邏輯。

3、觀察者模式通用代碼

/**
 * 觀察者
 */
public interface Observer {
    // 更新方法
    void update();
}

/**
 * 具體觀察者
 */
public class ConcreteObserver implements Observer{
    @Override
    public void update() {
        System.out.println("接受到信息，并進(jìn)行處理");
    }
}

/**
 * 被觀察者
 */
public abstract class Subject {
    // 定義一個(gè)被觀察者數(shù)組
    private List< Observer > obsList = new ArrayList<  >();

    // 增加一個(gè)觀察者
    public void addObserver(Observer observer){
        obsList.add(observer);
    }

    // 刪除一個(gè)觀察者
    public void delObserver(Observer observer){
        obsList.remove(observer);
    }

    // 通知所有觀察者
    public void notifyObservers(){
        for (Observer observer : obsList){
            observer.update();
        }
    }
}

/**
 * 具體被觀察者
 */
public class ConcreteSubject extends Subject{
    // 具體的業(yè)務(wù)
    public void doSomething(){
        super.notifyObservers();
    }
}

public class ObserverClient {

    public static void main(String[] args) {
        // 創(chuàng)建一個(gè)被觀察者
        ConcreteSubject subject = new ConcreteSubject();
        // 定義一個(gè)觀察者
        Observer observer = new ConcreteObserver();
        // 觀察者觀察被觀察者
        subject.addObserver(observer);
        subject.doSomething();
    }
}

4、JDK 實(shí)現(xiàn)

在 JDK 的 java.util 包下，已經(jīng)為我們提供了觀察者模式的抽象實(shí)現(xiàn),感興趣的可以看看，內(nèi)部邏輯其實(shí)和我們上面介紹的差不多。

觀察者 java.util.Observer

被觀察者 java.util.Observable

5、實(shí)例

用戶進(jìn)行注冊，注冊完成之后，會發(fā)一封歡迎郵件。

5.1 普通實(shí)現(xiàn)

public class UserController {

    public void register(String userName, String passWord){
        // 1、根據(jù)用戶名密碼保存在數(shù)據(jù)庫
        Long userId = saveUser(userName, passWord);
        // 2、如果上一步有結(jié)果則發(fā)送一封歡迎郵件
        if(userId != null){
            Mail.sendEmail(userId);
        }
    }


    public Long saveUser(String userName, String passWord){
        return 1L;
    }
}

上面的注冊接口實(shí)現(xiàn)了兩件事，注冊和發(fā)送郵件，很明顯違反了單一職責(zé)原則，但假設(shè)這個(gè)注冊需求是不是經(jīng)常變動的，這樣寫也沒有什么問題，但是假如需求變動，比如不僅要發(fā)送郵件，還得發(fā)送短信，那還這樣寫，那register接口會變得很復(fù)雜。

那應(yīng)該如何簡化呢？沒錯(cuò)，就是觀察者模式。

5.2 觀察者模式實(shí)現(xiàn)

我們直接套用 JDK 的實(shí)現(xiàn)。

import java.util.Observable;

/**
 * 用戶登錄——被觀察者
 */
public class UserControllerObservable extends Observable {

    public void register(String userName, String passWord){
        // 1、根據(jù)用戶名密碼保存在數(shù)據(jù)庫
        Long userId = saveUser(userName, passWord);
        // 2、如果上一步有結(jié)果則通知所有觀察者
        if(userId != null){
            super.setChanged();
            super.notifyObservers(userName);
        }
    }

    public Long saveUser(String userName, String passWord){
        return 1L;
    }

}

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

/**
 * 發(fā)送郵件——觀察者
 */
public class MailObserver implements Observer {

    @Override
    public void update(Observable o, Object arg) {
        System.out.println("發(fā)送郵件:" + arg + "歡迎你");
    }
}

/**
 * 發(fā)送手機(jī)短信——觀察者
 */
public class SMSObserver implements Observer {

    @Override
    public void update(Observable o, Object arg) {
        System.out.println("發(fā)送短信：" + arg + "歡迎你");
    }
}

測試：

public class UserClient {
    public static void main(String[] args) {
        UserControllerObservable observable = new UserControllerObservable();
        observable.addObserver(new MailObserver());
        observable.addObserver(new SMSObserver());
        observable.register("張三","123");
    }
}

通過觀察者模式改寫后，后面用戶注冊，就算在增加別的操作，我們也只需要增加一個(gè)觀察者即可，而注冊接口 register 不會有任何改動。

5.3 異步模式優(yōu)化

在回到前面那張圖：

注冊之后進(jìn)行的兩步操作：發(fā)送郵件和發(fā)送短信，上面我們通過觀察者模式改寫之后，雖然流程很清晰，但是我們發(fā)現(xiàn)是順序執(zhí)行的，但其實(shí)這兩步操作沒有先后順序，于是，我們可以改成異步模式，增加執(zhí)行效率。

/**
 * 發(fā)送郵件——觀察者
 */
public class MailObserver implements Observer {
    
    private Executor executor = Executors.newFixedThreadPool(2);

    @Override
    public void update(Observable o, Object arg) {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println("發(fā)送郵件:" + arg + "歡迎你");
            }
        });

    }
}

5、EventBus

翻譯為“事件總線”，它提供了實(shí)現(xiàn)觀察者模式的骨架代碼。我們可以基于此框架，非常容易地在自己的業(yè)務(wù)場景中實(shí)現(xiàn)觀察者模式，不需要從零開始開發(fā)。其中，Google Guava EventBus 就是一個(gè)比較著名的 EventBus 框架，它不僅僅支持異步非阻塞模式，同時(shí)也支持同步阻塞模式。

PS：Google Guava 是一個(gè)特別好用的工具包，里面的代碼也都實(shí)現(xiàn)的比較優(yōu)雅，大家感興趣的可以研究研究源碼。

https://github.com/google/guava

下面我們以上面的例子來說明如何使用 EventBus:

①、導(dǎo)如 Guava 包

< dependency >
    < groupId >com.google.guava< /groupId >
    < artifactId >guava< /artifactId >
    < version >30.1.1-jre< /version >
< /dependency >

②、具體代碼如下：

import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;

import java.util.List;
import java.util.concurrent.Executors;

public class UserController {
    private EventBus eventBus;

    public UserController(){
        eventBus = new AsyncEventBus(Executors.newFixedThreadPool(2));
    }

    /**
     * 注意：泛型參數(shù)是 Object,而不是接口 Observer
     * @param observerList
     */
    public void setObserverList(List< Object > observerList){
        for(Object observer : observerList){
            eventBus.register(observer);
        }
    }

    public void register(String userName, String passWord){
        // 1、根據(jù)用戶名密碼保存在數(shù)據(jù)庫
        Long userId = saveUser(userName, passWord);
        // 2、如果上一步有結(jié)果則通知所有觀察者
        if(userId != null){
            eventBus.post(userName);
        }
    }


    public Long saveUser(String userName, String passWord){
        return 1L;
    }
}

import com.google.common.eventbus.Subscribe;

/**
 * 發(fā)送郵件——觀察者
 */
public class MailObserver{

    @Subscribe
    public void sendMail(String userName) {
        System.out.println("發(fā)送郵件:" + userName + "歡迎你");
    }
}

import com.google.common.eventbus.Subscribe;

/**
 * 發(fā)送手機(jī)短信——觀察者
 */
public class SMSObserver{

    @Subscribe
    public void sendSMS(String userName) {
        System.out.println("發(fā)送短信：" + userName + "歡迎你");
    }
}

測試：

public class EventBusClient {
    public static void main(String[] args) {
        UserController userController = new UserController();
        List< Object > observerList = new ArrayList<  >();
        observerList.add(new MailObserver());
        observerList.add(new SMSObserver());
        userController.setObserverList(observerList);
        userController.register("張三","123");
    }
}

利用 EventBus 框架實(shí)現(xiàn)的觀察者模式，跟從零開始編寫的觀察者模式相比，從大的流程上來說，實(shí)現(xiàn)思路大致一樣，都需要定義 Observer，并且通過 register() 函數(shù)注冊 Observer，也都需要通過調(diào)用某個(gè)函數(shù)（比如，EventBus 中的 post() 函數(shù)）來給 Observer 發(fā)送消息（在 EventBus 中消息被稱作事件 event）。但在實(shí)現(xiàn)細(xì)節(jié)方面，它們又有些區(qū)別?；?EventBus，我們不需要定義 Observer 接口，任意類型的對象都可以注冊到 EventBus 中，通過 @Subscribe 注解來標(biāo)明類中哪個(gè)函數(shù)可以接收被觀察者發(fā)送的消息。

6、觀察者模式優(yōu)點(diǎn)

①、觀察者和被觀察者之間是抽象耦合

不管是增加觀察者還是被觀察者都非常容易擴(kuò)展，在系統(tǒng)擴(kuò)展方面會得心應(yīng)手。

②、建立一套觸發(fā)機(jī)制

被觀察者變化引起觀察者自動變化。但是需要注意的是，一個(gè)被觀察者，多個(gè)觀察者，Java的消息通知默認(rèn)是順序執(zhí)行的，如果一個(gè)觀察者卡住，會導(dǎo)致整個(gè)流程卡住，這就是同步阻塞。

所以實(shí)際開發(fā)中沒有先后順序的考慮使用異步，異步非阻塞除了能夠?qū)崿F(xiàn)代碼解耦，還能充分利用硬件資源，提高代碼的執(zhí)行效率。

另外還有進(jìn)程間的觀察者模式，通?；谙㈥?duì)列來實(shí)現(xiàn)，用于實(shí)現(xiàn)不同進(jìn)程間的觀察者和被觀察者之間的交互。

7、觀察者模式應(yīng)用場景

①、關(guān)聯(lián)行為場景。

②、事件多級觸發(fā)場景。

③、跨系統(tǒng)的消息交換場景， 如消息隊(duì)列的處理機(jī)制。