java Active Object模式(下)

Active Object模式的评价与实现考量

Active Object模式通过将方法的调用与执行分离，实现了异步编程。有利于提高并发性，从而提高系统的吞吐率。

Active Object模式还有个好处是它可以将任务（MethodRequest）的提交（调用异步方法）和任务的执行策略（Execution Policy）分离。任务的执行策略被封装在Scheduler的实现类之内，因此它对外是不“可见”的，一旦需要变动也不会影响其它代码，降低了系统的耦合性。任务的执行策略可以反映以下一些问题：

采用什么顺序去执行任务，如FIFO、LIFO、或者基于任务中包含的信息所定的优先级？
多少个任务可以并发执行？
多少个任务可以被排队等待执行？
如果有任务由于系统过载被拒绝，此时哪个任务该被选中作为牺牲品，应用程序该如何被通知到？
任务执行前、执行后需要执行哪些操作？

这意味着，任务的执行顺序可以和任务的提交顺序不同，可以采用单线程也可以采用多线程去执行任务等等。

当然，好处的背后总是隐藏着代价，Active Object模式实现异步编程也有其代价。该模式的参与者有6个之多，其实现过程也包含了不少中间的处理：MethodRequest对象的生成、MethodRequest对象的移动（进出缓冲区）、MethodRequest对象的运行调度和线程上下文切换等。这些处理都有其空间和时间的代价。因此，Active Object模式适合于分解一个比较耗时的任务（如涉及I/O操作的任务）：将任务的发起和执行进行分离，以减少不必要的等待时间。

虽然模式的参与者较多，但正如本文案例的实现代码所展示的，其中大部分的参与者我们可以利用JDK自身提供的类来实现，以节省编码时间。如表1所示。

表 1. 使用JDK现有类实现Active Object的一些参与者

参与者名称	可以借用的JDK类	备注
Scheduler	Java Executor Framework中的java.util.concurrent.ExecutorService接口的相关实现类，如java.util.concurrent.ThreadPoolExecutor。	ExecutorService接口所定义的submit(Callable<T> task)方法相当于图2中的enqueue方法。
ActivationQueue	java.util.concurrent.LinkedBlockingQueue	若Scheduler采用java.util.concurrent.ThreadPoolExecutor，则java.util.concurrent.LinkedBlockingQueue实例作为ThreadPoolExecutor构造器的参数。
MethodRequest	java.util.concurrent.Callable接口的匿名实现类。	Callable接口比起Runnable接口的优势在于它定义的call方法有返回值，便于将该返回值传递给Future实例。
Future	java.util.concurrent.Future	ExecutorService接口所定义的submit(Callable<T> task)方法的返回值类型就是java.util.concurrent.Future。

错误隔离

错误隔离指一个任务的处理失败不影响其它任务的处理。每个MethodRequest实例可以看作一个任务。那么，Scheduler的实现类在执行MethodRequest时需要注意错误隔离。选用JDK中现成的类（如ThreadPoolExecutor）来实现Scheduler的一个好处就是这些类可能已经实现了错误隔离。而如果自己编写代码实现Scheduler，用单个Active Object工作线程逐一执行所有任务，则需要特别注意线程的run方法的异常处理，确保不会因为个别任务执行时遇到一些运行时异常而导致整个线程终止。如清单6的示例代码所示。

清单 6. 自己动手实现Scheduler的错误隔离示例代码

 1 public class CustomScheduler implements Runnable {
 2     private LinkedBlockingQueue<Runnable> activationQueue = 
 3         new LinkedBlockingQueue<Runnable>();
 4 
 5     @Override
 6     public void run() {
 7         dispatch();
 8     }
 9 
10     public <T> Future<T> enqueue(Callable<T> methodRequest) {
11         final FutureTask<T> task = new FutureTask<T>(methodRequest) {
12 
13             @Override
14             public void run() {
15                 try {
16                    super.run();
17                 //捕获所以可能抛出的对象，避免该任务运行失败而导致其所在的线程终止。    
18                 } catch (Throwable t) {
19                    this.setException(t);
20                 }
21             }
22 
23         };
24         
25         try {
26             activationQueue.put(task);
27         } catch (InterruptedException e) {
28             Thread.currentThread().interrupt();
29         }
30         return task;
31     }
32 
33     public void dispatch() {
34         while (true) {
35             Runnable methodRequest;
36             try {
37                 methodRequest = activationQueue.take();
38                 
39                 //防止个别任务执行失败导致线程终止的代码在run方法中
40                 methodRequest.run();
41             } catch (InterruptedException e) {
42                 // 处理该异常
43             }
44 
45         }
46     }
47 }

缓冲区监控

如果ActivationQueue是有界缓冲区，则对缓冲区的当前大小进行监控无论是对于运维还是测试来说都有其意义。从测试的角度来看，监控缓冲区有助于确定缓冲区容量的建议值（合理值）。清单3所示的代码，即是通过定时任务周期性地调用ThreadPoolExecutor的getQueue方法对缓冲区的大小进行监控。当然，在监控缓冲区的时候，往往只需要大致的值，因此在监控代码中要避免不必要的锁。

缓冲区饱和处理策略

当任务的提交速率大于任务的执行数率时，缓冲区可能逐渐积压到满。这时新提交的任务会被拒绝。无论是自己编写代码还是利用JDK现有类来实现Scheduler，对于缓冲区满时新任务提交失败，我们需要一个处理策略用于决定此时哪个任务会成为“牺牲品”。若使用ThreadPoolExecutor来实现Scheduler有个好处是它已经提供了几个缓冲区饱和处理策略的实现代码，应用代码可以直接调用。如清单3的代码所示，本文案例中我们选择了抛弃最老的任务作为处理策略。java.util.concurrent.RejectedExecutionHandler接口是ThreadPoolExecutor对缓冲区饱和处理策略的抽象，JDK中提供的具体实现如表2所示。

表 2. JDK提供的缓冲区饱和处理策略实现类

实现类	所实现的处理策略
ThreadPoolExecutor.AbortPolicy	直接抛出异常。
ThreadPoolExecutor.DiscardPolicy	放弃当前被拒绝的任务（而不抛出任何异常）。
ThreadPoolExecutor.DiscardOldestPolicy	将缓冲区中最老的任务放弃，然后重新尝试接纳被拒绝的任务。
ThreadPoolExecutor.CallerRunsPolicy	在任务的提交方线程中运行被拒绝的任务。

当然，对于ThreadPoolExecutor而言，其工作队列满不一定就意味着新提交的任务会被拒绝。当其最大线程池大小大于其核心线程池大小时，工作队列满的情况下，新提交的任务会用所有核心线程之外的新增线程来执行，直到工作线程数达到最大线程数时，新提交的任务会被拒绝。

Scheduler空闲工作线程清理

如果Scheduler采用多个工作线程（如采用ThreadPoolExecutor这样的线程池）来执行任务。则可能需要清理空闲的线程以节约资源。清单3的代码就是直接使用了ThreadPoolExecutor的现有功能，在初始化其实例时通过指定其构造器的第3、4个参数（ long keepAliveTime, TimeUnit unit），告诉ThreadPoolExecutor对于核心工作线程以外的线程若其已经空闲了指定时间，则将其清理掉。

可复用的Active Object模式实现

尽管利用JDK中的现成类可以极大地简化Active Object模式的实现。但如果需要频繁地在不同场景下使用Active Object模式，则需要一套更利于复用的代码，以节约编码的时间和使代码更加易于理解。清单7展示一段基于Java动态代理的可复用的Active Object模式的Proxy参与者的实现代码。

清单 7. 可复用的Active Object模式Proxy参与者实现

 1 import java.lang.reflect.InvocationHandler;
 2 import java.lang.reflect.InvocationTargetException;
 3 import java.lang.reflect.Method;
 4 import java.lang.reflect.Proxy;
 5 import java.util.concurrent.Callable;
 6 import java.util.concurrent.ExecutorService;
 7 import java.util.concurrent.Future;
 8 public abstract class ActiveObjectProxy {
 9 
10     private static class DispatchInvocationHandler implements InvocationHandler {
11         private final Object delegate;
12         private final ExecutorService scheduler;
13 
14         public DispatchInvocationHandler(Object delegate,
15             ExecutorService executorService) {
16             this.delegate = delegate;
17             this.scheduler = executorService;
18         }
19 
20         private String makeDelegateMethodName(final Method method,
21             final Object[] arg) {
22             String name = method.getName();
23             name = "do" + Character.toUpperCase(name.charAt(0)) 
24                     + name.substring(1);
25 
26             return name;
27         }
28 
29         @Override
30         public Object invoke(final Object proxy, final Method method,
31             final Object[] args) throws Throwable {
32 
33             Object returnValue = null;
34             final Object delegate = this.delegate;
35             final Method delegateMethod;
36             
37             //如果拦截到的被调用方法是异步方法，则将其转发到相应的doXXX方法
38             if (Future.class.isAssignableFrom(method.getReturnType())) {
39                 delegateMethod = delegate.getClass().getMethod(
40                     makeDelegateMethodName(method, args),
41                     method.getParameterTypes());
42 
43                 final ExecutorService scheduler = this.scheduler;
44                 
45                 Callable<Object> methodRequest = new Callable<Object>() {
46                     @Override
47                     public Object call() throws Exception {
48                         Object rv = null;
49 
50                         try {
51                           rv = delegateMethod.invoke(delegate, args);
52                         } catch (IllegalArgumentException e) {
53                             throw new Exception(e);
54                         } catch (IllegalAccessException e) {
55                             throw new Exception(e);
56                         } catch (InvocationTargetException e) {
57                             throw new Exception(e);
58                         }
59                         return rv;
60                     }
61                 };
62                 Future<Object> future = scheduler.submit(methodRequest);
63                 returnValue = future;
64 
65             } else {
66                 
67                 //若拦截到的方法调用不是异步方法，则直接转发
68             delegateMethod = delegate.getClass()
69             .getMethod(method.getName(),method.getParameterTypes());
70 
71                 returnValue = delegateMethod.invoke(delegate, args);
72             }
73 
74             return returnValue;
75         }
76     }
77 
78     /**
79      * 生成一个实现指定接口的Active Object proxy实例。
80      * 对interf所定义的异步方法的调用会被装发到servant的相应doXXX方法。
81      * @param interf 要实现的Active Object接口
82      * @param servant Active Object的Servant参与者实例
83      * @param scheduler Active Object的Scheduler参与者实例
84      * @return Active Object的Proxy参与者实例
85      */
86     public static <T> T newInstance(Class<T> interf, Object servant,
87         ExecutorService scheduler) {
88 
89         @SuppressWarnings("unchecked")
90         T f = (T) Proxy.newProxyInstance(interf.getClassLoader(),
91         new Class[] { interf }, 
92         new DispatchInvocationHandler(servant, scheduler));
93 
94         return f;
95     }
96 }

清单7的代码实现了可复用的Active Object模式的Proxy参与者ActiveObjectProxy。ActiveObjectProxy通过使用Java动态代理，动态生成指定接口的代理对象。对该代理对象的异步方法（即返回值类型为java.util.concurrent.Future的方法）的调用会被ActiveObjectProxy实现InvocationHandler（DispatchInvocationHandler）所拦截，并转发给ActiveObjectProxy的newInstance方法中指定的Servant处理。

清单8所示的代码展示了通过使用ActiveObjectProxy快速Active Object模式。

清单 8. 基于可复用的API快速实现Active Object模式

1 public static void main(String[] args) throws 
2     InterruptedException, ExecutionException {
3 
4     SampleActiveObject sao = ActiveObjectProxy.newInstance(
5             SampleActiveObject.class, new SampleActiveObjectImpl(),
6             Executors.newCachedThreadPool());
7     Future<String> ft = sao.process("Something", 1);
8     Thread.sleep(500);
9     System.out.println(ft.get());

从清单8的代码可见，利用可复用的Active Object模式Proxy实现，应用开发人员只要指定Active Object模式对外保留的接口（对应ActiveObjectProxy.newInstance方法的第1个参数），并提供一个该接口的实现类（对应ActiveObjectProxy.newInstance方法的第2个参数），再指定一个java.util.concurrent.ExecutorService实例（对应ActiveObjectProxy.newInstance方法的第3个参数）即可以实现Active Object模式。

总结

本文介绍了Active Object模式的意图及架构。并提供了一个实际的案例用于展示使用Java代码实现Active Object模式，在此基础上对该模式进行了评价并分享了在实际运用该模式时需要注意的事项。

参考资源

本文的源代码在线阅读：https://github.com/Viscent/JavaConcurrencyPattern/
维基百科Active Object模式词条：http://en.wikipedia.org/wiki/Active_object
Douglas C. Schmidt对Active Object模式的定义：http://www.laputan.org/pub/sag/act-obj.pdf。
Schmidt, Douglas et al. Pattern-Oriented Software Architecture Volume 2: Patterns for Concurrent and Networked Objects. Volume 2. Wiley, 2000
Java theory and practice: Decorating with dynamic proxies：http://www.ibm.com/developerworks/java/library/j-jtp08305/index.html