异步编程提供了一个非阻塞的，事件驱动的编程模型。 这种编程模型利用系统中多核执行任务来提供并行，因此提高了应用的吞吐率。Java异步编程通常需要使用Future，FutureTask和Callable，其中Future代表未来的某个结果，一般是向线程池提交任务时返回。
Future
示例代码：
public class PromiseTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executor = Executors.newFixedThreadPool(3);
List<Future<String>> futureList = new ArrayList<>();
for (int i = 0; i < 3; i++) {
int finalI = i;
Future<String> future = executor.submit(() -> {
System.out.println("task["+ finalI +"] started!");
Thread.sleep(1000*(3-finalI));// cost some time
System.out.println("task["+ finalI +"]finished!");
return "result["+ finalI +"]";
});
futureList.add(future);
}
for (Future<String> future : futureList) {
System.out.println(future.get());
}
System.out.println("Main thread finished.");
executor.shutdown();
}
}
执行结果：
task[0] started!
task[2] started!
task[1] started!
task[2]finished!
task[1]finished!
task[0]finished!
result[0]
result[1]
result[2]
Main thread finished.
有两个问题值得注意：
- 向线程池提交的三个任务同时开始执行，但是在使用get取结果的时候发现必须等耗时最长的任务结束之后才可以得到执行结果。
- get方法阻塞了主线程，在取异步任务执行结果期间主线程不可以做其他事情。
这和真正意义上的异步编程模型是违背的，异步编程不存在线程阻塞的情况，因此仅使用Future来完成异步编程是不科学的。
CompletionService
CompletionService提供了一种将生产新的异步任务与使用已完成任务的结果分离开来的服务。生产者 submit 执行的任务，使用者 take 已完成的任务，并按照完成这些任务的顺序处理它们的结果，使用者总是可以取到最先完成任务的结果。
public class PromiseTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executor = Executors.newFixedThreadPool(3);
CompletionService<String> service = new ExecutorCompletionService<String>(executor);
for (int i = 0; i < 3; i++) {
int finalI = i;
service.submit(() -> {
System.out.println("task["+ finalI +"] started!");
Thread.sleep(1000*(3-finalI));// cost some time
System.out.println("task["+ finalI +"]finished!");
return "result["+ finalI +"]";
});
}
for (int i = 0; i < 3; i++) {
System.out.println(service.take().get());
}
System.out.println("Main thread finished.");
executor.shutdown();
}
}
执行结果：
task[0] started!
task[2] started!
task[1] started!
task[2]finished!
result[2]
task[1]finished!
result[1]
task[0]finished!
result[0]
Main thread finished.
可见，最耗时的任务在执行时并不影响其他主线程取到已完成任务的结果。但主线程仍然被get方法阻塞。
CompletableFuture
在JDK8发布之前，许多追求性能的框架都自己造了一套异步的实现，例如guava的ListenableFuture和netty的FutureListener，一直到Java8，才有了真正的官方异步实现， CompletableFuture提供了异步操作的支持，使用方法与Js中的Promise类似。
关于CompletableFuture的详细说明，请参考这篇：Java 8：CompletableFuture终极指南
使用这个API改写一下代码：
public class PromiseTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executor = Executors.newFixedThreadPool(3);
for (int i = 0; i < 3; i++) {
int finalI = i;
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
System.out.println("task["+finalI+"] started!");
try {
// time cost
Thread.sleep(1000*(3-finalI));
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("task["+finalI+"] finished");
return "result["+finalI+"]";
}, executor);
future.thenAccept(System.out::println);
}
System.out.println("Main thread finished.");
executor.shutdown();
}
}
结果与预期完全一致，各异步任务独自执行，互相不阻塞。
task[0] started!
Main thread finished.
task[1] started!
task[2] started!
task[2] finished
result[2]
task[1] finished
result[1]
task[0] finished
result[0]
异步IO
从编程模型角度来讲，IO操作方式有BIO/NIO/AIO，BIO就是传统的阻塞IO模型，关于NIO的介绍，也请参考这篇： Java NIO基础
与NIO区别的是，API提供的读写方法均是异步的，即对流或通道内容进行读写时并不会阻塞调用的主线程，JDK7在java.nio.channels包新增了以下4个API来支持异步IO：
- AsynchronousSocketChannel
- AsynchronousServerSocketChannel
- AsynchronousFileChannel
- AsynchronousDatagramChannel
AIO并没有加速IO处理速度，只是利用回调和通知机制改变了业务处理时机，使得具体逻辑可以不关注IO结果，只需在合理的时机添加回调即可。
例如：
public class AsyncServer {
public static void main(String[] args) {
try {
AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(8000));
server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
final ByteBuffer buffer = ByteBuffer.allocate(1024);
@Override
public void completed(AsynchronousSocketChannel result, Object attachment) {
Future<Integer> writeResult = null;
try {
buffer.clear();
result.read(buffer).get(100, TimeUnit.SECONDS);
buffer.flip();
writeResult = result.write(buffer);
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
server.accept(null, this);
writeResult.get();
result.close();
} catch (Exception e) {
System.out.println(e.toString());
}
}
}
@Override
public void failed(Throwable exc, Object attachment) {
}
});
} catch (IOException e) {
e.printStackTrace();
}
}
}
通过CompletionHandler来添加一些回调，这样可以很方便地进行异步IO操作。
网友评论