[[403490]]

消息隊(duì)列(MQ)，一種能實(shí)現(xiàn)生產(chǎn)者到消費(fèi)者單向通信的通信模型，這也是現(xiàn)在常用的主流中間件。

常見有 RabbitMQ、ActiveMQ、Kafka等 他們的特點(diǎn)也有很多 比如 解偶、異步、廣播、削峰 等等多種優(yōu)勢(shì)特點(diǎn)。

在設(shè)計(jì)模式中也有一種模式能有效的達(dá)到解偶、異步的特點(diǎn)，那就是觀察者模式又稱為發(fā)布訂閱模式。

今天阿丙就分享一下實(shí)際開發(fā)中比較常見的這種模式

大綱

定義

什么是觀察者模式?他的目的是什么?

• 當(dāng)一個(gè)對(duì)象的狀態(tài)發(fā)生改變時(shí)，已經(jīng)登記的其他對(duì)象能夠觀察到這一改變從而作出自己相對(duì)應(yīng)的改變。通過這種方式來達(dá)到減少依賴關(guān)系，解耦合的作用。

舉一個(gè)例子，就好比微信朋友圈，以當(dāng)前個(gè)人作為訂閱者，好友作為主題。一個(gè)人發(fā)一條動(dòng)態(tài)朋友圈出去，他的好友都能看到這個(gè)朋友圈，并且可以在自主選擇點(diǎn)贊或者評(píng)論。

感覺有點(diǎn)抽象，還是看看他有哪些主要角色：

• Subject(主題): 主要由類實(shí)現(xiàn)的可觀察的接口，通知觀察者使用attach方法，以及取消觀察的detach方法。
• ConcreteSubject(具體主題): 是一個(gè)實(shí)現(xiàn)主題接口的類，處理觀察者的變化
• Observe(觀察者): 觀察者是一個(gè)由對(duì)象水岸的接口，根據(jù)主題中的更改而進(jìn)行更新。

這么看角色也不多，但是感覺還是有點(diǎn)抽象，我們還是用具體實(shí)例代碼來走一遍吧，我們還是以上面的朋友圈為例看看代碼實(shí)現(xiàn)

public interface Subject { 
    // 添加訂閱關(guān)系 
    void attach(Observer observer); 
    // 移除訂閱關(guān)系 
    void detach(Observer observer); 
    // 通知訂閱者 
    void notifyObservers(String message); 
} 

先創(chuàng)建一個(gè)主題定義，定義添加刪除關(guān)系以及通知訂閱者

public class ConcreteSubject implements Subject { 
 
    // 訂閱者容器 
    private List<Observer> observers = new ArrayList<Observer>(); 
 
    @Override 
    public void attach(Observer observer) { 
        // 添加訂閱關(guān)系 
        observers.add(observer); 
    } 
 
    @Override 
    public void detach(Observer observer) { 
        // 移除訂閱關(guān)系 
        observers.remove(observer); 
    } 
 
    @Override 
    public void notifyObservers(String message) { 
        // 通知訂閱者們 
        for (Observer observer : observers) { 
            observer.update(message); 
        } 
    } 
} 

其次再創(chuàng)建的具體主題，并且構(gòu)建一個(gè)容器來維護(hù)訂閱關(guān)系，支持添加刪除關(guān)系，以及通知訂閱者

public interface Observer { 
    // 處理業(yè)務(wù)邏輯 
    void update(String message); 
} 

創(chuàng)建一個(gè)觀察者接口，方便我們管理

public class FriendOneObserver implements Observer { 
    
  @Override 
    public void update(String message) { 
        // 模擬處理業(yè)務(wù)邏輯 
        System.out.println("FriendOne 知道了你發(fā)動(dòng)態(tài)了" + message); 
    } 
} 

最后就是創(chuàng)建具體的觀察者類，實(shí)現(xiàn)觀察者接口的update方法，處理本身的業(yè)務(wù)邏輯

public class test { 
     
    public static void main(String[] args) { 
 
        ConcreteSubject subject = new ConcreteSubject(); 
        // 這里假設(shè)是添加好友 
        subject.attach(new FriendOneObserver()); 
        FriendTwoObserver twoObserver = new FriendTwoObserver(); 
        subject.attach(twoObserver); 
 
        // 發(fā)送朋友圈動(dòng)態(tài) 
        subject.notifyObservers("第一個(gè)朋友圈消息"); 
        // 輸出結(jié)果：FriendOne 知道了你發(fā)動(dòng)態(tài)了第一個(gè)朋友圈消息 
        //          FriendTwo 知道了你發(fā)動(dòng)態(tài)了第一個(gè)朋友圈消息 
 
        // 這里發(fā)現(xiàn) twoObserver 是個(gè)推薦賣茶葉的，刪除好友 
        subject.detach(twoObserver); 
        subject.notifyObservers("第二個(gè)朋友圈消息"); 
        // 輸出結(jié)果：FriendOne 知道了你發(fā)動(dòng)態(tài)了第二個(gè)朋友圈消息 
    } 
} 

最后就是看測(cè)試結(jié)果了，通過ConcreteSubject 維護(hù)了一個(gè)訂閱關(guān)系，在通過notifyObservers 方法通知訂閱者之后，觀察者都獲取到消息從而處理自己的業(yè)務(wù)邏輯。

這里細(xì)心的朋友已經(jīng)達(dá)到了解耦合的效果，同時(shí)也減少了依賴關(guān)系，每個(gè)觀察者根本不要知道發(fā)布者處理了什么業(yè)務(wù)邏輯，也不用依賴發(fā)布者任何業(yè)務(wù)模型，只關(guān)心自己本身需要處理的邏輯就可以了。

如果有新的業(yè)務(wù)添加進(jìn)來，我們也只需要?jiǎng)?chuàng)建一個(gè)新的訂閱者，并且維護(hù)到observers 容器中即可，也符合我們的開閉原則。

這里只是一種同步的實(shí)現(xiàn)方式，我們還可以擴(kuò)展更多其他的異步實(shí)現(xiàn)方式，或者采用多線程等實(shí)現(xiàn)方式。

框架應(yīng)用

觀察者模式在框架的中的應(yīng)用也是應(yīng)該很多

• 第一種 熟悉JDK的人應(yīng)該知道 在java.util 包下 除了常用的 集合 和map之外還有一個(gè)Observable類，他的實(shí)現(xiàn)方式其實(shí)就是觀察者模式。里面也有添加、刪除、通知等方法。

這里需要注意是的 他是用Vector 作為訂閱關(guān)系的容器，同時(shí)在他的定義方法中都添加synchronized關(guān)鍵字修飾類，以達(dá)到線程安全的目的

這里我貼出了關(guān)鍵源碼，感興趣的同學(xué)可以自己打開并且觀看每個(gè)方法的注釋。

• 第二種 在Spring中有一個(gè)ApplicationListener，也是采用觀察者模式來處理的，ApplicationEventMulticaster作為主題，里面有添加，刪除，通知等。

spring有一些內(nèi)置的事件，當(dāng)完成某種操作時(shí)會(huì)發(fā)出某些事件動(dòng)作，他的處理方式也就上面的這種模式，當(dāng)然這里面還有很多，我沒有細(xì)講，有興趣的同學(xué)可以仔細(xì)了解下Spring的啟動(dòng)過程。

import java.util.EventListener;/** * Interface to be implemented by application event listeners. * Based on the standard {@code java.util.EventListener} interface * for the Observer design pattern. // 這里也已經(jīng)說明是采用觀察者模式 * *

import java.util.EventListener; 
 
/** 
 * Interface to be implemented by application event listeners. 
 * Based on the standard {@code java.util.EventListener} interface 
 *  for the Observer design pattern. // 這里也已經(jīng)說明是采用觀察者模式 
 * 
 * <p>As of Spring 3.0, an ApplicationListener can generically declare the event type 
 * that it is interested in. When registered with a Spring ApplicationContext, events 
 * will be filtered accordingly, with the listener getting invoked for matching event 
 * objects only. 
 * 
 * @author Rod Johnson 
 * @author Juergen Hoeller 
 * @param <E> the specific ApplicationEvent subclass to listen to 
 * @see org.springframework.context.event.ApplicationEventMulticaster //主題 
 */ 
@FunctionalInterface 
public interface ApplicationListener<E extends ApplicationEvent> extends EventListener { 
 
 /** 
  * Handle an application event. 
  * @param event the event to respond to 
  */ 
 void onApplicationEvent(E event); 
 
} 

• 第三種 Google Guava的事件處理機(jī)制Guava EventBus 他的實(shí)現(xiàn)也是采用設(shè)計(jì)模式中的觀察者設(shè)計(jì)模式。

EventBus 當(dāng)前實(shí)現(xiàn)有兩種方式：

• EventBus // 同步阻塞模式
• AsyncEventBus // // 異步非阻塞模式

EventBus內(nèi)部也提供來一系列的方法來供我們方便使用：

• register 方法作為添加觀察者
• unregister方法刪除觀察者
• post 方法發(fā)送通知消息等

使用起來非常方便。添加@Subscribe注解就可以創(chuàng)建一個(gè)訂閱者了，具體的使用方式可以看看官網(wǎng)。

現(xiàn)實(shí)業(yè)務(wù)改造舉例

框架應(yīng)用的例子這么多，在業(yè)務(wù)場(chǎng)景中其實(shí)也有很多地方可以使用到，這里我還是給大家舉一個(gè)例子。

在新用戶注冊(cè)成功之后我們需要給用戶做兩件事情，第一是發(fā)送注冊(cè)成功短信，第二是給用發(fā)送新人優(yōu)惠券。

看到這個(gè)問題 大家可能首先會(huì)想到用MQ消息處理呀，是的，用消息確實(shí)可以的，但是這里我們用觀察者模式來實(shí)現(xiàn)這個(gè)問題，同時(shí)可以給大家演示一下，同步或者異步的問題。

public class SendNewPersonCouponObserver implements Observer { 
 
    ExecutorService pool = Executors.newFixedThreadPool(2); 
 
    @Override 
    public void update(String message) { 
 
        Future<String> future = pool.submit(new Callable<String>() { 
            @Override 
            public String call() throws Exception { 
                TimeUnit.SECONDS.sleep(3); 
                // 處理響應(yīng)的業(yè)務(wù)邏輯 
                return "調(diào)用發(fā)券服務(wù)，返回結(jié)果"; 
            } 
        }); 
        try { 
            // 假設(shè)等待200毫秒 沒有獲取到返回值結(jié)果則認(rèn)為失敗 
            System.out.println(future.get(4000, TimeUnit.MILLISECONDS)); 
        } catch (Exception e) { 
            // 執(zhí)行異步獲取失敗 
            // 記錄日志，定時(shí)任務(wù)重試等 
        } 
 
        // 第一種不關(guān)心返回值結(jié)果 
        Thread thread = new Thread(new Runnable() { 
            @SneakyThrows 
            @Override 
            public void run() { 
                // 模擬服務(wù)調(diào)用 線程睡3秒鐘 
                TimeUnit.SECONDS.sleep(3); 
                System.out.println("發(fā)送新人優(yōu)惠券"); 
            } 
        }); 
        thread.start(); 
        System.out.println("執(zhí)行異步返回"); 
    } 
} 

public class SendSuccessMessageObserver implements Observer { 
 
    @Override 
    public void update(String message) { 
        // 處理業(yè)務(wù)邏輯 
        System.out.println("注冊(cè)成功"); 
    } 
 
    public static void main(String[] args) { 
        // 假設(shè)用戶注冊(cè)成功直接通知觀察者，改干自己的事情了 
        ConcreteSubject subject = buildSubject(); 
        subject.notifyObservers(""); 
    } 
  
   private static ConcreteSubject buildSubject() { 
        ConcreteSubject subject = new ConcreteSubject(); 
        subject.attach(new SendSuccessMessageObserver()); 
        subject.attach(new SendNewPersonCouponObserver()); 
        return subject; 
    } 
} 

這里我們新寫了兩個(gè)觀察者，主要看第一個(gè)SendNewPersonCouponObserver，這里了異步開啟新的線程去處理我們的業(yè)務(wù)邏輯，當(dāng)我們關(guān)心返回值的時(shí)候可以用Future來獲取返回結(jié)果，當(dāng)不關(guān)心的返回值的化，直接開啟普通線程就可以了。

這個(gè)舉例整體其實(shí)還是比較簡(jiǎn)單的主要是為了說清楚異步線程處理，當(dāng)然如果用Guava EventBus也可以實(shí)現(xiàn)。而且也不復(fù)雜，感興趣的朋友可以自己去試試。

當(dāng)前現(xiàn)在有更加好的中間件MQ消息隊(duì)列來處理這個(gè)業(yè)務(wù)問題，使得我們更加從容的面對(duì)這類場(chǎng)景問題，但是一些資源不足，不想引入新的系統(tǒng)。還是可以用這種方式來處理問題的。

設(shè)計(jì)模式學(xué)習(xí)的不是代碼，而是學(xué)習(xí)每種模式的思想，他們分別處理的是什么業(yè)務(wù)場(chǎng)景。

總結(jié)

大家看完本篇文章不知道有發(fā)現(xiàn)沒有，其實(shí)整個(gè)內(nèi)容都是圍繞了解耦的思想來寫的，觀察者模式作為行為型設(shè)計(jì)模式，主要也是為了不同的業(yè)務(wù)行為的代碼解耦。

合理的使用設(shè)計(jì)模式可以使代碼結(jié)構(gòu)更加清晰，同時(shí)還能滿足不同的小模塊符合單一職責(zé)，以及開閉原則，從而達(dá)到前面寫工廠模式說的，提高代碼的可擴(kuò)展性，維護(hù)成本低的特點(diǎn)。

我是敖丙你知道的越多，你不知道的越多，我們下期見。