異步編程提供了一個非阻塞的续滋，事件驅(qū)動的編程模型忘伞。 這種編程模型利用系統(tǒng)中多核執(zhí)行任務(wù)來提供并行库糠，因此提高了應(yīng)用的吞吐率。Java異步編程通常需要使用Future剥纷，FutureTask和Callable痹籍，其中Future代表未來的某個結(jié)果，一般是向線程池提交任務(wù)時返回晦鞋。
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();
}
}
執(zhí)行結(jié)果：
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.
有兩個問題值得注意：
- 向線程池提交的三個任務(wù)同時開始執(zhí)行蹲缠，但是在使用get取結(jié)果的時候發(fā)現(xiàn)必須等耗時最長的任務(wù)結(jié)束之后才可以得到執(zhí)行結(jié)果。
- get方法阻塞了主線程悠垛，在取異步任務(wù)執(zhí)行結(jié)果期間主線程不可以做其他事情线定。
這和真正意義上的異步編程模型是違背的，異步編程不存在線程阻塞的情況确买，因此僅使用Future來完成異步編程是不科學(xué)的斤讥。
CompletionService
CompletionService提供了一種將生產(chǎn)新的異步任務(wù)與使用已完成任務(wù)的結(jié)果分離開來的服務(wù)。生產(chǎn)者 submit 執(zhí)行的任務(wù)湾趾，使用者 take 已完成的任務(wù)芭商，并按照完成這些任務(wù)的順序處理它們的結(jié)果，使用者總是可以取到最先完成任務(wù)的結(jié)果搀缠。
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();
}
}
執(zhí)行結(jié)果：
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.
可見铛楣，最耗時的任務(wù)在執(zhí)行時并不影響其他主線程取到已完成任務(wù)的結(jié)果。但主線程仍然被get方法阻塞胡嘿。
CompletableFuture
在JDK8發(fā)布之前蛉艾，許多追求性能的框架都自己造了一套異步的實現(xiàn)，例如guava的ListenableFuture和netty的FutureListener，一直到Java8勿侯，才有了真正的官方異步實現(xiàn)拓瞪， CompletableFuture提供了異步操作的支持，使用方法與Js中的Promise類似助琐。
關(guān)于CompletableFuture的詳細(xì)說明祭埂，請參考這篇：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();
}
}
結(jié)果與預(yù)期完全一致，各異步任務(wù)獨自執(zhí)行兵钮，互相不阻塞蛆橡。
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就是傳統(tǒng)的阻塞IO模型泰演，關(guān)于NIO的介紹，也請參考這篇： Java NIO基礎(chǔ)
與NIO區(qū)別的是葱轩，API提供的讀寫方法均是異步的睦焕，即對流或通道內(nèi)容進(jìn)行讀寫時并不會阻塞調(diào)用的主線程，JDK7在java.nio.channels包新增了以下4個API來支持異步IO：
- AsynchronousSocketChannel
- AsynchronousServerSocketChannel
- AsynchronousFileChannel
- AsynchronousDatagramChannel
AIO并沒有加速IO處理速度靴拱，只是利用回調(diào)和通知機制改變了業(yè)務(wù)處理時機垃喊，使得具體邏輯可以不關(guān)注IO結(jié)果，只需在合理的時機添加回調(diào)即可袜炕。
例如：
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來添加一些回調(diào)本谜，這樣可以很方便地進(jìn)行異步IO操作。
