## 1. 前言
在并发编程中,异步回调的效率不言而喻,在业务开发中,如果由阻塞的任务需要执行,必然要使用异步线程。并且,如果我们想在异步执行之后,根据他的结果执行一些动作。
JDK 8 之前的 Future 只能解决上面需求的一半问题,即异步执行,返回一个 Future,需要程序员调用 get 方法等待,或者使用 isDone 轮询。
效率不高。
JDK 8 新出的 CompletableFuture API 可以解决这个问题。但他的 API, 说实话,不太好用。
我们只想要一个简单的 API,能实现我们的回调功能。
我需要 3 个功能:
- 能通过 get 之类的方法返回结果。
- 能设置监听器进行回调。
- 可以在业务线程中设置成功或者失败。
楼主写一个简单的例子,借鉴了 Netty 的异步 API,希望能起到抛砖引玉的作用。
2. 设计
根据我们的需求:
第一,我们需要一个类,拥有 get 方法和 addListener 方法。
第二,我们需要一个类,能够回调我们设置的监听器。
第三,我们需要一个类,能够在业务线程中设置成功或者失败。
3. 初步实现
设计一个监听器接口:
1 2 3 4 5 6 7 8 9 10 11
|
public interface MyListener {
void operationComplete(MyPromise promise); }
|
设计一个异步占位符,类似 Future:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90
|
public class MyPromise {
List<MyListener> listeners = new ArrayList<MyListener>();
boolean success;
Object result;
int failCount;
public boolean setSuccess(Object result) { if (success) { return false; }
success = true; this.result = result;
signalListeners(); return true; }
private void signalListeners() { for (MyListener l : listeners) { l.operationComplete(this); } }
public boolean setFail(Exception e) { if (failCount > 0) { return false; } ++failCount; result = e; signalListeners(); return true; }
public boolean isSuccess() { return success; }
public void addListener(MyListener myListener) { listeners.add(myListener); }
public void removeListener(MyListener myListener) { listeners.remove(myListener); }
public Object get() { return result; } }
|
我们希望使用线程池执行此类任务,所以需要一个自定义的 Runnable,而在这个 Runnable 中,我们需要做一些简单的手脚:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
|
public abstract class MyRunnable<V> implements Runnable {
final MyPromise myPromise;
protected MyRunnable(MyPromise myPromise) { this.myPromise = myPromise; }
@Override public void run() { try { V v = doWork(); myPromise.setSuccess(v); } catch (Exception e) { myPromise.setFail(e); } }
public abstract V doWork(); }
|
4. 写个 Demo 测试一下
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54
|
public class MyDemo {
public static void main(String[] args) {
final MyPromise myPromise = new MyPromise();
final Dto dto = new Dto();
Executor executor = Executors.newFixedThreadPool(1);
executor.execute(new MyRunnable<String>(myPromise) { @Override public String doWork() { return dto.doSomething(); } });
myPromise.addListener(new MyListener() { @Override public void operationComplete(MyPromise promise) { String result; if (promise.isSuccess()) { result = (String) promise.get(); System.out.println("operationComplete ----> " + result); } else { ((Exception) promise.get()).printStackTrace(); } } }); }
}
class Dto {
public String doSomething() { System.out.println("doSomething");
return "result is success"; } }
|
执行结果:
1 2
| doSomething operationComplete ----> result is success
|
符合我们的预期。我们希望在业务对象 Dto 的 doSomething 成功返回之后,回调监听器的 operationComplete 方法。如果失败,打印异常堆栈。
当然,整体代码比较简单,仅仅只是抛砖引玉。
实际上,如果直接向 Callable 或者 Runnable 传入一个业务对象,当 call 方法或者 run 方法执行完毕,就可以根据执行结果执行我们的业务对象的方法了。这样就是一个最简单直接的异步回调。
只是这样过于耦合。
异步任务和业务的任务耦合在了一起,并且不能添加多个监听器,也无法使用 promise 的 setSuccess 功能和 setFail 功能,这两个功能可以在业务线程中设置成功或者失败,灵活性更高。
关于异步,我们可以多看看 Netty 的 API 设计,易懂好用。