如何实现异步计算？

Question

我一直在考虑用Java实现异步处理的不同方法。我想出了几个可能性，想听听你对其中一些是否比其他的更好的看法，也许还能得到关于如何做到这一点的进一步建议。想到的最常见的用例是通过下面的API、void sendData(Data data)甚至void sendData(Data data, Handler handler)通过连接(例如TCP)发送数据包。以下是我的一些想法：

1. 专用数据发送循环-有一个didicated线程，它的行为有点像Java中的事件调度线程，所有其他线程都调用并提交一个请求。这些请求被存储在一个队列中，并周期性地释放所有请求。
2. 使用后台线程来清除队列--连接可以维护挂起的请求列表，并有一个后台线程来清除队列(与某些同步)。
3. Executor服务-将请求传递给服务并从后台线程调用Handler。
4. 异步通道-最高级的方法，委托给实现。

是否有更好的使用，或如果您有任何其他的一般性想法，请分享您的意见。

Answer 1

你的大部分建议实际上是不同的方式来打扮相同的事情。

在后台，executor服务有一个由一个或多个后台线程组成的池，用于排走队列。向它提交了请求。

1和2都是“排队处理事情，有一个线程来处理它”的方式。

基本上，1和2都是彼此的变体。它们也是3的子集。

我不知道你在这里是什么意思？

ExecutorService的设计就是要做您想要做的事情--如果您有多个线程来执行任务，那么它显然就是执行任务的方法。

如果您只有一个线程，您仍然可以使用ExecutorService，但选择不太清楚。仅仅使用一个线程和一个BlockingQueue就相当简单了，所以这可能是可行的。

Answer 2

我总是使用执行器服务来执行并发。它们的高度足以隐藏管理线程的复杂性，并允许线程重用。您可以将任务提交给执行器，执行器将有效地充当队列，或者有许多任务运行时使用同步队列(如阻塞队列)来共享数据。后者可能允许更大的灵活性，例如批处理队列项。

我还强烈推荐Guava ListenableFuture，因为它可能解决您在使用并发时可能遇到的许多问题。

Answer 3

您应该明确地将异步管道设计为具有数据/事件依赖关系的图形。典型图由两种节点组成：

fast handler:
     immediately invoked when an event happens and:
     probably stores the event,
     and/or calls another fast handler,
     and/or submits an asynchronous task to an executor

task:
     runs and issues events (that is, calls fast handlers)

因此，基本上您需要开发两个独立的层:快速处理程序和任务执行器。执行器是通用的，可以从java.util.concurrent包中提取。快速处理程序在很大程度上取决于您的问题域，并且没有针对所有情况的通用库。例如，纯队列是一个只存储事件的快速处理程序，因此几乎毫无用处。

如果您使用I/O，则需要使用标准I/O库，以便它们为您的处理程序发出I/O事件。它可以使用用于同步I/O的线程构建，也可以使用Selector线程或异步I/O的异步通道构建。

与NIO2异步通道一起使用的快速处理程序示例：

 class ConnectionAcceptor  implements CompletionHandler<AsynchronousSocketChannel, Void>{
   AsynchronousServerSocketChannel assc;
   int maxConn;// max number of simultaneous connections
   int connCount=0;

   /* called on creation to start listening incoming client connection requests */
   void allowAccept() {
       assc.accept(null, this);
   }

   /* called by I/O layer when a client connection requested */
   public synchronized void completed(AsynchronousSocketChannel result, Void attachment) {
      executor.exec(new Connection(result));
      connCount++;
      if (connCount<maxConn) {
        allowAccept();
      }
   }

   /* called by Connection when it is closed */
   synchronized void connClosed() {
      if (connCount==maxConn) {
        allowAccept();
      }
      connCount--;
   }
}

这个处理程序有3个入口点，并处理3种事件。类似地，可以创建读或写(但不是两者都)的处理程序。它的内部状态不包含connCount，而是包含一个布尔标志，指示I/O操作正在进行中，以及等待AsynchronousSocketChannel结束已启动操作的缓冲区队列。