多线程观察者设计模式

　　假设我们要关心一些事物发生的状态 那么我们可能就需要不断的对这个事物的状态进行查询 而这显然是很低效的 最好的方式应该是当这个事物发生变化时 它会通知我们 而不是我们一味的去查询 这显然会节省大量的时间 这种通知方式就是观察者的设计模式，当某个对象发生状态改变需要通知第三方的时候，观察者模式就特别适合胜任这样的工作 下面我们先看一个比较简单的例子

　　假设有一个订阅号 每当它发布新文章的时候 它就会通知所有订阅过的用户 用户当收到文章若不喜欢 它可以退订 此后就收不到这个订阅号更新的信息了，这个事件观察者设计模式是很好胜任的，订阅号就相当于是事件源，它有新的推送就相当于状态发生变化，然后就会通知有订阅的观察者，即用户。要实现这个例子，首先显然需要设计一个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中定义的方法，它相当于事件回调的响应者