Active Object 并发模式在 Java 中的应用

简介：?Active Object 是并发编程实践中典型的设计模式，Active Object 模式的核心是通过解耦合方法的调用与执行来提高程序的并发度。本文将从典型 Active Object 设计模式入手，从一个新的视角来探讨 Active Object 并发模式在 Java 中的应用。

?

本文主要从以下两个方面进行阐述:

使用 C++ 语言，来描述 Active Object 设计模式。

Java 类库对于这样一个典型的模式做了很好的类库层面的封装，因此对于 Java 的开发者来说，很多关于该设计模式本身的东西被屏蔽掉了。本文试图使用 Native C++ 语言，帮助读者从本质上对 Active Object 设计模式有一个更全面的认识。

结合 C++ 版本的 Active Object 设计模式，引领读者对于 Active Object 设计模式在 Java 类库中的支持，有一个更深刻的认识，帮助读者正确并且有效地使用这些类库。

从图 1 我们可以看到，步骤 1 到步骤 6 运行在调用者线程中，而步骤 7 到步骤 12 运行在 Active Object 的线程中。

同一个进程中的多个调用者线程可以共享同一个 Proxy。

实现 Method Request，如清单 3 所示：

清单 3. Method_Request

Activation List 的实际上就是一个线程同步机制保护下的 Method Request 队列，对该队列的所有操作 (insert/remove) 都应该是线程安全的。从本质上讲，Activation List 所基于的就是典型的生产者 / 消费者并发编程模型，调用者线程作为生产者把 Method Request 放入该队列，Active Object 线程作为消费者从该队列拿出 Method Request, 并执行。

实现 Scheduler，如清单 5 所示：

清单 5. MQ_Scheduler

class MQ_Scheduler {public:    // Initialize the <Activation_List> and make <MQ_Scheduler>    // run in its own thread of control.         // we call this thread as Active Object thread.    MQ_Scheduler ()    : act_list_() {        // Spawn separate thread to dispatch method requests.        // The following call is leveraging the parallelism available on native OS        // transparently        Thread_Manager::instance ()->spawn (&svc_run,this);    }    // ... Other constructors/destructors, etc.    // Put <Method_Request> into <Activation_List>. This    // method runs in the thread of its client,i.e.    // in the proxy's thread.    void insert (Method_Request *mr) {        act_list_.insert (mr);    }    // Dispatch the method requests on their servant    // in its scheduler's thread of control.    virtual void dispatch () {        // Iterate continuously in a separate thread(Active Object thread).        for (;;) {            Activation_List::iterator request;            // The iterator's <begin> method blocks            // when the <Activation_List> is empty.            for(request = act_list_.begin (); request != act_list_.end ();++request){                // Select a method request whose                // guard evaluates to true.                if ((*request).can_run ()) {                    // Take <request> off the list.                    act_list_.remove (*request);                    (*request).call () ;                    delete *request;                }                // Other scheduling activities can go here,                // e.g., to handle when no <Method_Request>s                // in the <Activation_List> have <can_run>                // methods that evaluate to true.            }        }    }private:    // List of pending Method_Requests.    Activation_List act_list_;    // Entry point into the new thread.    static void *svc_run (void *arg) {        MQ_Scheduler *this_obj =    static_cast<MQ_Scheduler *> (args);        this_obj->dispatch ();    }};

实现 Future，如清单 6 所示：

清单 6. Message_Future

class Message_Future {public:    // Initializes <Message_Future> to    // point to <message> immediately.    Message_Future (const Message &message);    //Other implementatio……    // Block upto <timeout> time waiting to obtain result    // of an asynchronous method invocation. Throws    // <System_Ex> exception if <timeout> expires.    Message result (Time_Value *timeout = 0) const;private:    //members definition here……};

事实上，对于调用者来说，可以通过以下的方式从 Future 对象获得真实的执行结果 Message:

同步等待。调用者调用 Future 对象的 result() 方法同步等待，直到后端的 Servant 相应方法执行结束，并把结果存储到了 Future 对象中来，result 返回，调用者获得 Message。同步超时等待。调用者调用 Future 对象的 result(timeout) 方法。如果过了 timeout 时间之后，后端的 Servant 相应方法执行仍未结束，则调用失败，否则，调用者线程被唤醒，result 方法返回，调用者获得 Message。异步查询。调用者可以通过调用 Future 对象定义的查询方法 ( 清单 6 没有提供相应的定义 )，查看真实的结果是否准备好了，如果准备好了，调用 result 方法，直接获得 Message。

清单 7 是使用该 Active Object 的示例。

清单 7. Active Object 使用

MQ_Proxy message_queue;//Optioin 1. Obtain future and block thread until message arrives.Message_Future future = message_queue.get();Message msg = future.result();//Handle received message herehandle(msg);//2. Obtain a future (does not block the client).Message_Future future = message_queue.get ();//The current thread is not blocked, do something else here...//Evaluate future and block if result is not available.Message msg = future.result ();//Handle received message herehandle(msg);

从清单 7 可以看到，MQ_Proxy 对于调用者而言，和一个普通的 C++ 定义的对象并没有区别，并发的实现细节已经被隐藏。

Java 对 Active Object 支持

Java JDK 1.3 引入了 java.util.Timer 和 java.util.TimerTask，提供了对 timer-based 并发任务支持，Timer 和 TimerTask 可以看作是 Active Object 设计模式在 Java 中的实现。不过，在这里我们不打算过多讨论 Timer 及其 TimerTask。由于 Timer 和 TimerTask 的缺陷性，例如 Timer 使用单线程执行 TimerTask 导致的 Task 调度时间的不精确性等问题。从 Java1.5 开始，Java 建议使用 ScheduledThreadPoolExecutor 作为 Timer 的替代。

在这里，我们讨论一下自 Java1.5 引入的 Executor Framework。Java1.5 的 Executor Framework 可以看作是 Active Object 设计模式在 Java 中的体现。不过 Java 的 Executor Framework 极大地简化了我们前面所讨论的 Active Object 所定义的模式。

Java 的 Executor Framework 是一套灵活强大的异步任务执行框架，它提供了标准的方式解耦合任务的提交与任务的执行。Java Executor 框架中的任务指的是实现了 Runnable 或者 Callable 接口的对象。Executor 的示例用法如清单 8 所示：

清单 8. Java Executor 示例代码

public class TaskExecutionTcpServer {    private static final int NTHREADS = 100;    private static final Executor exec = Executors.newFixedThreadPool(NTHREADS);    public static void main(String[] args) throws IOException {        ServerSocket socket = new ServerSocket(80);        while (true) {            final Socket connection = socket.accept();            Runnable task = new Runnable() {                public void run() {                    handleRequest(connection);                }                public void handleRequest(Socket connection) {                    // Handle the incoming socket connection from                                             //individual client.                }            };            exec.execute(task);        }    }}

?

在示例 8 中，我们创建了一个基于线程池的 Java Executor, 每当新的 TCP 连接进来的时候，我们就分配一个独立的实现了 Runnable 任务来处理该连接，所有这些任务运行在我们创建的有 100 个线程的线程池上。

我们可以从 Active Object 设计模式的角度来审视一下 Java Executor 框架。Java Executor 框架以任务 (Task) 为中心，简化了 Active Object 中的角色分工。可以看到，实现 Runnable 或者 Callable 接口的 Java Executor 任务整合了 Method Request 和 Servant 的角色 , 通过实现 run() 或者 call() 方法实现应用逻辑。Java Executor 框架并没有显式地定义 Proxy 接口，而是直接调用 Executor 提交任务，这里的 Executor 相当于 Active Object 中调度者角色。从调用者的角度来看，这看起来并不像是在调用一个普通对象方法，而是向 Executor 提交了一个任务。所以，在这个层面上说，并发的底层细节已经暴露给了调用者。对于 Java 的开发者来说，如果你不担心这样的底层并发细节直接暴露给调用者，或者说你的应用并不需要像对待普通对象一样对待并发对象，Java 的 Executor 框架是一个很好的选择。相反，如果你希望隐藏这样的并发细节，希望像操纵普通对象一样操纵并发对象，那你就需要如本文上节所描述的那样，遵循 Active Object 设计原则 , 清晰地定义各个角色，实现自己的 Active Object 模式。

总而言之，Java Executor 框架简化了 Active Object 所定义的模式，模糊了 Active Object 中角色的分工，其基于生产者 / 消费者模式，生产者和消费者基于任务相互协作。

?

总结

最后，我们讨论一下 Active Object 设计模式的优缺点。

Active Object 给我们的应用带来的好处：

极大提高了应用的并发性以及简化了线程同步带来的复杂性。并发性的提高得益于调用者线程与 Active Object 线程的并发执行。简化的线程同步复杂性主要表现在所有线程同步细节封装在调度者内 ( 也就是 Java 的 Executor 对象 )，Active Object 调用者并不需要关心。在 Active Object 中，方法的执行顺序可以不同于方法的调用顺序。用 Java 的话说，也就是任务执行的顺序可以不同于任务提交的顺序。在一定情况下，这可以帮助优化我们应用的性能，提高应用的 QoS 及其 Responsiveness。在 Java Executor 框架下，你可以根据当前的计算资源，确定优化的执行策略 (Execution Policy)，该执行策略的内容包括：任务将分配在多少线程上执行，以什么顺序执行，多少任务可以同时执行等等。

当然，Active Object 也有缺点：

额外的性能开销。这涉及到从调用者线程到 Active Object 线程的上下文切换，线程同步，额外的内存拷贝等。难于调试。Active Object 引入了方法的异步执行，从调试者的角度看，调试这样的方法调用不像普通方法那样直截了当，并且这其中涉及到了线程的调度，同步等。

?

参考资料

学习

“JDK 5.0 中的并发”（developerWorks，2004 年 12 月）：本教程将介绍 JDK 5.0 提供的用于并发的新实用程序类，并通过与现有并发原语（synchronized、wait() 和 notify()）相比较，说明这些类如何提高了可伸缩性。

“Java 多线程与并发编程专题”（developerWorks，2008 年 6 月）：Java 平台提供了一套广泛而功能强大的 API、工具和技术。其中，内建支持线程是它的一个强大的功能。这一功能为使用 Java 编程语言的程序员提供了并发编程这一诱人但同时也非常具有挑战性的选择。本专题汇集了与 Java 多线程与并发编程相关的文章和教程，帮助读者理解 Java 并发编程的模式及其利弊，向读者展示了如何更精确地使用 Java 平台的线程模型。

“Java 设计模式与建模专题”（developerWorks，2008 年 1 月）：本专题为 Java 软件工程师们提供了面向 Java 的设计模式和建模方面相关的文章和教程。帮助读者理解、学习作为专业软件工程师必需掌握的设计模式与建模技术。

技术书店：浏览关于这些和其他技术主题的图书。

developerWorks Java 技术专区：数百篇关于 Java 编程各个方面的文章。

?

原文：http://www.ibm.com/developerworks/cn/java/j-lo-activeobject/index.html?ca=drs-

?

?

?

?

?

?