Java中的异步编程有`CompletionHandler`和`CompletableFuture`两种不同的方法吗？

Question

在Java中的异步I/O果壳

回调风格是基于CompletionHandler，这 定义了两种方法，completed()和failed()，将被称为回来的时候 操作要么成功或失败。 如果您希望立即通知异步事件中的事件 I/O（例如，如果有大量I/O操作在运行中，但任何单个操作的故障 未必是致命的），此样式非常有用。

从http://www.deadcoderising.com/java8-writing-asynchronous-code-with-completablefuture/

除了实现Future接口，CompletableFuture也 实现CompletionStage接口。

A CompletionStage是一个承诺。它承诺计算 最终将完成。

关于CompletionStage的好处是，它提供了大量的方法选择，允许您附加将在完成时执行 的回调。

这样我们就可以以非阻塞的方式构建系统。

CompletionHandler和CompletableFuture都可用于指定回叫处理程序。

他们的关系和差异是什么？

是CompletionHandler和CompletableFuture两种不同的用于Java中异步编程的方法吗？

或者他们一起使用？

谢谢。

Answer 1

如果你看看CompletableFuture（自Java 8以来），你会注意到它具有大量的功能，允许比回调更多的功能。与链接，组合和其他有趣的功能。

与CompletionHandler（自Java 7以来）相比，差别应该是显而易见的。

没有什么能够阻止你使用两者，根据你使用的API类型，甚至有必要使用它，但如果你有机会使用CompletableFuture，那么你确实不需要使用CompletionHandler 。

Answer 2

CompletionHandler<V, A>是NIO异步通道的完成接口。在Java 8引入lambda表达式并将它们转换为函数接口（使用单一抽象方法的接口）之前的几年，Java 7引入了它，所以它有两种方法，即completed(V result, A attachment)和failed(Throwable ex, A attachment)，而不是（现在）更舒适的单一方法。

CompletableFuture<T>是CompletionStage<T>接口的实现。

它是在Java 8中引入的，所以它的静态方法来构建期货及其实例方法来构建延续使用函数接口，您可以使用它来舒适地使用lambda表达式。

你可以用每NIO的异步调用使用CompletableFuture<T>：

import java.lang.*; 
import java.net.*; 
import java.nio.*; 
import java.nio.channels.*; 
import java.util.*; 
import java.util.concurrent.*; 

public class AsynchronousCompletionHandler<T> implements CompletionHandler<T, CompletableFuture<T>> { 
    public void completed(T result, CompletableFuture<T> attachment) { 
     attachment.complete(result); 
    } 

    public void failed(Throwable ex, CompletableFuture<T> attachment) { 
     attachment.completeExceptionally(ex); 
    } 

    private static final ConcurrentHashMap<Class<?>, AsynchronousCompletionHandler<?>> cache = new ConcurrentHashMap<>(); 

    static <T> AsynchronousCompletionHandler<T> getInstance(Class<T> clazz) { 
     @SuppressWarnings("unchecked") 
     AsynchronousCompletionHandler<T> handler = (AsynchronousCompletionHandler<T>)cache.computeIfAbsent(clazz, c -> new AsynchronousCompletionHandler<T>()); 
     return handler; 
    } 

    // 
    // AsynchronousByteChannel 
    public static CompletableFuture<Integer> readAsync(AsynchronousByteChannel channel, ByteBuffer dst) { 
     CompletableFuture<Integer> completableFuture = new CompletableFuture<>(); 
     channel.read(dst, completableFuture, getInstance(Integer.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<Integer> writeAsync(AsynchronousByteChannel channel, ByteBuffer src) { 
     CompletableFuture<Integer> completableFuture = new CompletableFuture<>(); 
     channel.write(src, completableFuture, getInstance(Integer.class)); 
     return completableFuture; 
    } 

    // 
    // AsynchronousFileChannel 
    public static CompletableFuture<FileLock> lockAsync(AsynchronousFileChannel channel) { 
     CompletableFuture<FileLock> completableFuture = new CompletableFuture<>(); 
     channel.lock(completableFuture, getInstance(FileLock.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<FileLock> lockAsync(AsynchronousFileChannel channel, long position, long size, boolean shared) { 
     CompletableFuture<FileLock> completableFuture = new CompletableFuture<>(); 
     channel.lock(position, size, shared, completableFuture, getInstance(FileLock.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<Integer> readAsync(AsynchronousFileChannel channel, ByteBuffer dst, long position) { 
     CompletableFuture<Integer> completableFuture = new CompletableFuture<>(); 
     channel.read(dst, position, completableFuture, getInstance(Integer.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<Integer> writeAsync(AsynchronousFileChannel channel, ByteBuffer src, long position) { 
     CompletableFuture<Integer> completableFuture = new CompletableFuture<>(); 
     channel.write(src, position, completableFuture, getInstance(Integer.class)); 
     return completableFuture; 
    } 

    // 
    // AsynchronousServerSocketChannel 
    public static CompletableFuture<AsynchronousSocketChannel> acceptAsync(AsynchronousServerSocketChannel channel) { 
     CompletableFuture<AsynchronousSocketChannel> completableFuture = new CompletableFuture<>(); 
     channel.accept(completableFuture, getInstance(AsynchronousSocketChannel.class)); 
     return completableFuture; 
    } 

    // 
    // AsynchronousSocketChannel 
    public static CompletableFuture<Void> connectAsync(AsynchronousSocketChannel channel, SocketAddress remote) { 
     CompletableFuture<Void> completableFuture = new CompletableFuture<>(); 
     channel.connect(remote, completableFuture, getInstance(Void.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<Long> readAsync(AsynchronousSocketChannel channel, ByteBuffer[] dsts, int offset, int length, long timeout, TimeUnit unit) { 
     CompletableFuture<Long> completableFuture = new CompletableFuture<>(); 
     channel.read(dsts, offset, length, timeout, unit, completableFuture, getInstance(Long.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<Integer> readAsync(AsynchronousSocketChannel channel, ByteBuffer dst) { 
     CompletableFuture<Integer> completableFuture = new CompletableFuture<>(); 
     channel.read(dst, completableFuture, getInstance(Integer.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<Integer> readAsync(AsynchronousSocketChannel channel, ByteBuffer dst, long timeout, TimeUnit unit) { 
     CompletableFuture<Integer> completableFuture = new CompletableFuture<>(); 
     channel.read(dst, timeout, unit, completableFuture, getInstance(Integer.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<Long> writeAsync(AsynchronousSocketChannel channel, ByteBuffer[] srcs, int offset, int length, long timeout, TimeUnit unit) { 
     CompletableFuture<Long> completableFuture = new CompletableFuture<>(); 
     channel.write(srcs, offset, length, timeout, unit, completableFuture, getInstance(Long.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<Integer> writeAsync(AsynchronousSocketChannel channel, ByteBuffer src) { 
     CompletableFuture<Integer> completableFuture = new CompletableFuture<>(); 
     channel.write(src, completableFuture, getInstance(Integer.class)); 
     return completableFuture; 
    } 

    public static CompletableFuture<Integer> writeAsync(AsynchronousSocketChannel channel, ByteBuffer src, long timeout, TimeUnit unit) { 
     CompletableFuture<Integer> completableFuture = new CompletableFuture<>(); 
     channel.write(src, timeout, unit, completableFuture, getInstance(Integer.class)); 
     return completableFuture; 
    } 
} 

样例用法（仅用于阐述;没有错误处理，不处理资源）：

import static AsynchronousCompletionHandler; 

AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open(); 
serverChannel.bind(new InetSocketAddress(5000)); 
ByteBuffer buffer = ByteBuffer.allocateDirect(1024); 
acceptAsync(serverChannel) 
    .thenCompose(clientChannel -> readAsync(clientChannel, buffer)) 
    .thenAccept(readBytes -> System.out.format("read %d bytes from client%n", readBytes));