多線程觀察者設計模式

　　假設我們要關心一些事物發生的狀態 那么我們可能就需要不斷的對這個事物的狀態進行查詢 而這顯然是很低效的 最好的方式應該是當這個事物發生變化時 它會通知我們 而不是我們一味的去查詢 這顯然會節省大量的時間 這種通知方式就是觀察者的設計模式，當某個對象發生狀態改變需要通知第三方的時候，觀察者模式就特別適合勝任這樣的工作 下面我們先看一個比較簡單的例子

　　假設有一個訂閱號 每當它發布新文章的時候 它就會通知所有訂閱過的用戶 用戶當收到文章若不喜歡 它可以退訂 此后就收不到這個訂閱號更新的信息了，這個事件觀察者設計模式是很好勝任的，訂閱號就相當於是事件源，它有新的推送就相當於狀態發生變化，然后就會通知有訂閱的觀察者，即用戶。要實現這個例子，首先顯然需要設計一個Observer的集合 它就是承載用戶的集合 它就是觀察者 當這個訂閱號發布文章的時候 就會向這些觀察者更新信息 

　　首先先定義一個Observer的接口 里面只有一個方法就是更新 給觀察者實現 即給用戶實現 

public interface Observer { void update(); }

　　然后定義一個ObserversWay的接口 它是訂閱號實現的 訂閱號顯然是清楚整個事物的狀態發生的變化 可以向各個觀察者發送更新的信息，用戶退訂和訂閱即為移除觀察者和添加觀察者

public interface ObserversWay { void add(Observer observer); //添加觀察者 void remove(Observer observer);//移除觀察者，即移除用戶 void notifyAllObservers();//更新信息 }

　　接下來是訂閱號的具體實現

public class Subject implements ObserversWay{
    List<Observer> Observers;  //定義一個觀察者的隊列 承載用戶 

    public Subject() {
        Observers = new ArrayList<>();
    }
    
    private String msg;

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        if (this.msg == msg) {
            return;
        }
        this.msg = msg;
        notifyAllObservers();  //當發表新文章時更新信息
    }

    @Override
    public void add(Observer observer) { //添加用戶
        this.Observers.add(observer);
    }

    @Override
    public void remove(Observer observer) {　 //用戶退訂
        this.Observers.remove(observer);
    }


    @Override
    public void notifyAllObservers() {
        for (Observer observer : Observers) {   //向所有用戶發送更新的信息
            observer.update();
        }
    }
}

接下來是用戶的實現

public class User implements Observer { private final String name; public User(String name) { this.name = name; } @Override public void update(String message) { System.out.println(this.name + ":has received " + message); //發表新消息時觸發的方法 } }

接下來是測試

public static void main(String[] args) { Subject subject = new Subject(); User user1 = new User("lakersFans A"); User user2 = new User("CelticFans"); User user3 = new User("lakersFans B"); //用戶1、2、3訂閱了該訂閱號
 subject.add(user1); subject.add(user2); subject.add(user3); subject.setMsg("湖人總冠軍"); //凱爾特人球迷看到這條推送后果斷退訂
 subject.remove(user2); System.out.println("======================="); subject.setMsg("凱爾特人總冠軍"); //這條推送凱爾特人粉絲就看不見了
    }

結果：

lakersFans A:has received 湖人總冠軍 CelticFans:has received 湖人總冠軍 lakersFans B:has received 湖人總冠軍 ======================= lakersFans A:has received 凱爾特人總冠軍 lakersFans B:has received 凱爾特人總冠軍

 

在對觀察者模式有了初步的了解后，接下來我們來對更具體的場景進行分析 通過觀察者模式去實現如何對一個任務的生命周期進行觀察 首先需要定義一個枚舉類型

定義任務的生命周期狀態

public enum Cycle { STARTED,RUNNING,DONE,ERROR }

在定義好枚舉類型后 定義一個關於任務生命周期的接口 當這個任務發生狀態變化hi通知監控者時，監控者對應的執行方法 相當於上述例子中更新推送時會執行的update方法

public interface TaskLifeCycle<T> {
//這個任務開始時觸發的方法 void onStart(Thread thread); //這個任務運行時觸發的方法 void onRunning(Thread thread); //這個任務完成時觸發的方法 result為關心的返回值 void onFinish(Thread thread,T result); //出現異常時觸發的方法 void onError(Thread thread, Exception e); }

接下來定義一個調用者的接口 通過這個接口的實現 來實現整個監控流程的實現 被觀察的任務會實現這個接口 當它發生狀態變化時 就會觸發上述的方法

public interface Observable { //獲取當前任務的生命周期
 Cycle getCycle(); //啟動線程去監控 還有一個作用是屏蔽Thread的其他API
    void start(); //對當前監控線程進行打斷
    void interrupt(); }

最后定義一個返回值的接口

public interface Task<T> {
//主要是需要關心返回值時調用 T call(); }

接下來是調用者監控時的具體實現 通過這個方法 會對任務的生命周期進行監控 當任務的生命周期發生變化時 任務就會通知調用者 讓調用者執行

public final class ObservableThread<T> extends Thread implements Observable { 　　//定義的監控者實現 監控者根據狀態變化采取的行動 private final TaskLifeCycle<T> lifeCycle; 　　//返回值 private final Task<T> task; 　　//狀態 private Cycle cycle; public ObservableThread(TaskLifeCycle<T> lifeCycle, Task<T> task) { //Thread的構造方法
        super(); if (task == null) throw new IllegalArgumentException("Task is not allowed for null"); this.lifeCycle = lifeCycle; this.task = task; } 　　//更新狀態 private void update(Cycle cycle, T result, Exception e) { this.cycle = cycle; if (lifeCycle == null) { return; } try { switch (cycle) { case STARTED: this.lifeCycle.onStart(currentThread()); break; case RUNNING: this.lifeCycle.onRunning(currentThread()); break; case DONE: this.lifeCycle.onFinish(currentThread(), result); break; case ERROR: this.lifeCycle.onError(currentThread(), e); break; } } catch (Exception ex) { if (cycle == Cycle.ERROR) { throw ex; } } } 　　//被監控的任務的邏輯實現單元 它執行到哪一個階段就會觸發相應的方法，即監控者定義的方法 @Override public void run() { this.update(Cycle.STARTED, null, null); try {this.update(Cycle.RUNNING, null, null); TimeUnit.SECONDS.sleep(5); //睡眠5秒模擬工作 T result = this.task.call(); this.update(Cycle.DONE, result, null); } catch (Exception e) { this.update(Cycle.ERROR, null, e); //發生異常時觸發的方法 } } 　　//獲取任務的執行狀態 @Override public Cycle getCycle() { return this.cycle; } }

接下來簡單實現監控者的接口，設立發生狀態變化時需要執行的對應方法

public class lifeCycle<T> implements TaskLifeCycle<T> { @Override public void onStart(Thread thread) { System.out.println("i started"); } @Override public void onRunning(Thread thread) { System.out.println("i am working"); } @Override public void onFinish(Thread thread, T result) { System.out.println("i finished"); System.out.println(result); } @Override public void onError(Thread thread, Exception e) { System.out.println("i went wrong"); } }

接下來是測試 

public static void main(String[] args) { lifeCycle lifeCycle = new lifeCycle(); ObservableThread observableThread = new ObservableThread(lifeCycle, () -> "任務完成啦"); observableThread.start(); }

結果

i started
i am working
i finished
任務完成啦  //定義的返回值

如果在observableThread的run方法中添加一個1/0，結果如下

i started
i am working
i went wrong

總結：在這個監控任務生命周期的實現中 充當事件源的是ObservableThread，它的run方法就是任務的邏輯單元和事件的發起者，在執行過程中若發生狀態變化就會觸發監控者定義好的方法，即TaskLifeCycle中定義的方法，它相當於事件回調的響應者