Java高并发编程详解系列-线程生命周期观察者

引言

在之前的博客中我们知道，Thread提供了很多可获取的状态，以及判断是否alive的方法，但是这些方法都是线程本身提供的，在Runnable运行的过程中所处的状态是无法直接获取到的到，例如什么时候开始的？什么时候结束的？都需要另外的方法进行获取。有的时候为了监听线程的状态不得不再其中设置共享变量。但是在多线程的情况下设置共享变量导致资源竞争从而增加了其他数据不一致的安全隐患。

在学习设计模式的时候我们假设了如下的场景，当一个某个对象的变化要被其他线程或者对象知道的时候，我们就是使用观察者模式。在这里，对于线程的监控也可以使用观察者模式。当然观察者模式需要有观察事件的来源，也就是导致事件变化的源头。在多线程的情况下很明显Thread类就作为线程事件改变的源头，这个源头在整个生命周期中事件的变化要通知事件接收方，也就是说需要一个观察者来监视事件的变化。下面就通过一个简单的例子来看一下关于这些模式的使用

定义一个接口

public interface Observable {//任务生命周期的枚举类型enum Cycle{STARED,RUNNING,DONE,ERROR}//获取当前任务生命周期状态Cycle getCycle();//定义启动线程的方法，主要是为了隐藏Thread的其他方法void start();//定义线程终止的方法，作用和start方法一样，避免Thread的方法void interrupt();
}

这个接口的主要作用是暴露给调用者使用。设置了四个枚举类型分别代表当前任务的执行生命周期。

getCycle()方法用于获取当前的任务在生命周期的哪个阶段
start()方法主要是为了屏蔽Thread的其他的方法，可以通过这个方法来启动整个线程
interrupt()方法的作用和start的方法作用相同

TaskLifecycle接口定义

public interface TaskLifecycle<T> {//任务启动时会触发onStart方法void onStart(Thread thread);//任务运行的时候触发onRunning方法void onRunning(Thread thread);//任务结束的时候会触发onFinish方法，其中result是任务执行完成之后的结果void onFinish(Thread thread,T result);//任务出错的时候会触发onError方法void onError(Thread thread,Exception e);//生命周期接口的空实现 Adapterclass EmptyLifecycle<T> implements TaskLifecycle<T>{@Overridepublic void onStart(Thread thread) {}@Overridepublic void onRunning(Thread thread) {}@Overridepublic void onFinish(Thread thread, T result) {}@Overridepublic void onError(Thread thread, Exception e) {}}
}

这个接口主要是定义了在线程执行过程中会被触发的那些接口，其中我们看到有一个空实现内部类。其中

onStart()方法表示任务开始执行的时候会被调用的方法
onRunning()方法表示在任务执行过程中会被调用的方法，这个也是根据线程的生命周期来定义的。
onFinish()方法表示整个任务顺利执行完成之后会调用的方法。
onError()方法表示在运行的任何时间内出现异常都会被这个方法回调。

Task函数接口定义

@FunctionalInterface
public interface Task<T> {//任务执行接口，该接口允许有返回值T call();
}

由于需要对线程中执行的任务做监听，需要实现一个回调方法，将结果进行返回所以实现了这样的一个接口。

ObservableThread类实现

public class ObservableThread<T> extends Thread implements Observable {private final TaskLifecycle<T> lifecycle;private final Task<T> task;private Cycle cycle;public ObservableThread(Task<T> task) {this(new TaskLifecycle.EmptyLifecycle<>(),task);}public ObservableThread(TaskLifecycle<T> lifecycle,Task<T> task) {super();if (task==null){throw new IllegalArgumentException("The task is required.");}this.lifecycle = lifecycle;this.task = task;}public final void run(){this.update(Cycle.STARED,null,null);try {this.update(Cycle.RUNNING,null,null);T result = this.task.call();this.update(Cycle.DONE,result,null);}catch (Exception e){this.update(Cycle.ERROR,null,e);}}private void update(Cycle cycle,T result,Exception e){this.cycle = cycle;if (lifecycle==null){return;}try{switch (cycle){case STARED: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;}}}@Overridepublic Cycle getCycle() {return this.cycle;}
}

在这里作为整个监控任务的核心，整个类继承了Thread类和Observable接口，并且在构造期间需要传入任务的具体实现。然后重写了父类的run()方法并且将其设置为final，我们知道final表示整个方法不会再被重写。所以说在生命周期的任何时间内每发生一个事件都要低啊用update方法，这个方法就是需要通知监听者任务执行过程中的变化。其实这里的原理和call()方法的原理是一样的，当任务状态发生改变的时候就要让观察者及时更新。

总结

通过实现我们会发现给他的实现其实与Thread的实现并没有多大的区别，只是ObservableThread作为一个泛型存在，因为返回值为Void也就表示并不在乎他的返回值是什么。

public class Main {public static void main(String[] args) {Observable observableThread = new ObservableThread<>(()->{try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("finish done");return null;});observableThread.start();}
}

在下面的实现中我们会发现其实对于EmptyLifecycle也是没有返回值。但是通过onFinish方法将最后的结果进行的返回。

public class Test {public static void main(String[] args) {final TaskLifecycle<String> lifecycle = new TaskLifecycle.EmptyLifecycle<String>() {@Overridepublic void onFinish(Thread thread, String result) {System.out.println("The result is " + result);}};Observable observableThread = new ObservableThread<>(lifecycle,()->{try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(" finished done.");return "Hello Observer";});observableThread.start();}}

关键

在上面的实现过程中作为核心的一点就是将Thread的方法进行了封装。首先屏蔽了Thread的API，其次封装了run方法让其不能被继承，为了实现状态的及时通知提供了两个方法call和update。但是最主要的就是离不开封装和展示。封装了不愿意被用户看到的，展示了想观察者所想要的。Observable类作为观察者最想看到的东西暴露出来。在后续博客中讲解分享设计模式的时候还会对这些内容进行总结，敬请期待!