回顾总结多线程-观察者模式在多线程中的使用

Author: sleepxdq

design-patterns/multithreading/doc.md

@@ -0,0 +1,3 @@
+多线程架构设计
+---
+* [监控任务的生命周期](/observer/doc.md)

design-patterns/multithreading/observer/build.gradle.kts

@@ -0,0 +1,7 @@
+plugins {
+ java
+}
+
+group = "com.xdq.observer"
+version = "0.0.1"
+java.sourceCompatibility = JavaVersion.VERSION_1_8

design-patterns/multithreading/observer/doc.md

@@ -0,0 +1,15 @@
+监控任务的生命周期
+---
+### 场景
+在使用多线程时，我们执行完start后其实不会等待线程的执行结果，
+有时我们会用同步等待的方式去获取线程的执行结果，这样其实是不好的。
+所以我们可以不去使用主动的方式获取执行结果，而是在执行结束后被告知结果，
+这样及可以准时的获取执行结果反馈，也不需要阻塞式的获取线程执行结果。
+### 模型
+>使用观察者模型
+
+![任务周期模型](http://assets.processon.com/chart_image/608adcd7e0b34d47a7146064.png)
+
+### 实践
+1. 非Spring容器下使用，见[代码示例](/src/main/java/com/xdq/observer/Main.java)
+2. 如何在Spring容器中使用

design-patterns/multithreading/observer/src/main/java/com/xdq/observer/Cycle.java

@@ -0,0 +1,9 @@
+package com.xdq.observer;
+
+/**
+ * @author Seven.Xu
+ * @version 2021/4/30
+ **/
+public enum Cycle {
+ STARTED, RUNNING, DONE, ERROR
+}

design-patterns/multithreading/observer/src/main/java/com/xdq/observer/EmptyLifecycle.java

@@ -0,0 +1,27 @@
+package com.xdq.observer;
+
+/**
+ * @author Seven.Xu
+ * @version 2021/4/30
+ **/
+public class EmptyLifecycle<T> implements TaskLifecycle<T> {
+ @Override
+ public void onStart(Thread thread) {
+
+ }
+
+ @Override
+ public void onRunning(Thread thread) {
+
+ }
+
+ @Override
+ public void onFinish(Thread thread, T result) {
+
+ }
+
+ @Override
+ public void onError(Thread thread, Exception e) {
+
+ }
+}

design-patterns/multithreading/observer/src/main/java/com/xdq/observer/Main.java

@@ -0,0 +1,40 @@
+package com.xdq.observer;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author Seven.Xu
+ * @version 2021/4/30
+ **/
+public class Main {
+ public static void main(String[] args) {
+ TaskLifecycle<String> lifecycle = new TaskLifecycle<String>() {
+ @Override
+ public void onStart(Thread thread) {
+ System.out.println("任务start");
+ }
+
+ @Override
+ public void onRunning(Thread thread) {
+ System.out.println("任务running");
+ }
+
+ @Override
+ public void onFinish(Thread thread, String result) {
+ System.out.println("任务finish,result:" + result);
+ }
+
+ @Override
+ public void onError(Thread thread, Exception e) {
+ System.out.println("任务error,exception:" + e);
+ }
+ };
+
+ Observable observable = new ObservableThread<>(lifecycle, () -> {
+ TimeUnit.SECONDS.sleep(2);
+ System.out.println("任务内容");
+ return "任务结果";
+ });
+ observable.start();
+ }
+}

design-patterns/multithreading/observer/src/main/java/com/xdq/observer/Observable.java

@@ -0,0 +1,25 @@
+package com.xdq.observer;
+
+/**
+ * 需要Thread其他方法可以在这个接口里扩展
+ * @author Seven.Xu
+ * @version 2021/4/30
+ **/
+public interface Observable {
+ /**
+ * 获取当前任务的生命周期
+ * @return
+ */
+ Cycle getCycle();
+
+ /**
+ * 定义启动线程的方法，主要作用为了屏蔽Thread的其他方法
+ */
+ void start();
+
+ /**
+ * 定义打断线程的方法，也是为了屏蔽Thread的方法
+ */
+ void interrupt();
+
+}

design-patterns/multithreading/observer/src/main/java/com/xdq/observer/ObservableThread.java

@@ -0,0 +1,79 @@
+package com.xdq.observer;
+
+import java.util.Objects;
+
+/**
+ * @author Seven.Xu
+ * @version 2021/4/30
+ **/
+public class ObservableThread<R> extends Thread implements Observable {
+
+ private final TaskLifecycle<R> lifecycle;
+ private final Task<R> task;
+ private Cycle cycle;
+
+ /**
+ * 指定Task实现，默认执行EmptyLifecycle
+ *
+ * @param task
+ */
+ public ObservableThread(Task<R> task) {
+ this(new TaskLifecycle.EmptyLifecycle<R>(), task);
+ }
+
+ public ObservableThread(TaskLifecycle<R> lifecycle, Task<R> task) {
+ super();//初始化Thread
+ Objects.requireNonNull(task,"task cannot be null");
+ this.lifecycle = lifecycle;
+ this.task = task;
+ }
+
+ @Override
+ public Cycle getCycle() {
+ return this.cycle;
+ }
+
+ /**
+ * final防止重写
+ */
+ @Override
+ public final void run() {
+ this.update(Cycle.STARTED, null, null);
+ try {
+ this.update(Cycle.RUNNING, null, null);
+ R result = this.task.call();
+ this.update(Cycle.DONE, result, null);
+ } catch (Exception e) {
+ this.update(Cycle.ERROR, null, e);
+ }
+ }
+
+ private void update(Cycle cycle, R result, Exception e) {
+ this.cycle = cycle;
+ if (lifecycle == null) {
+ return;
+ }
+ try {
+ switch (this.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);
+ }
+ } catch (Exception ex) {
+ //不能屏蔽任务执行的异常
+ if (cycle == Cycle.ERROR) {
+ throw ex;
+ }
+ //通知后出现异常需要catch，防止影响任务的正常执行
+ }
+
+ }
+}

design-patterns/multithreading/observer/src/main/java/com/xdq/observer/Task.java

@@ -0,0 +1,15 @@
+package com.xdq.observer;
+
+/**
+ * @author Seven.Xu
+ * @version 2021/4/30
+ **/
+@FunctionalInterface
+public interface Task<R> {
+
+ /**
+ * 任务执行接口
+ * @return
+ */
+ R call() throws InterruptedException;
+}

design-patterns/multithreading/observer/src/main/java/com/xdq/observer/TaskLifecycle.java

@@ -0,0 +1,38 @@
+package com.xdq.observer;
+
+/**
+ * @author Seven.Xu
+ * @version 2021/4/30
+ **/
+public interface TaskLifecycle<T> {
+ void onStart(Thread thread);
+
+ void onRunning(Thread thread);
+
+ void onFinish(Thread thread, T result);
+
+ void onError(Thread thread,Exception e);
+
+ class EmptyLifecycle<T> implements TaskLifecycle<T>{
+
+ @Override
+ public void onStart(Thread thread) {
+
+ }
+
+ @Override
+ public void onRunning(Thread thread) {
+
+ }
+
+ @Override
+ public void onFinish(Thread thread, T result) {
+
+ }
+
+ @Override
+ public void onError(Thread thread, Exception e) {
+
+ }
+ }
+}