Future和CompletableFuture区别

2023-08-21 22:15 由 xietingweia 发表于 #后端开发

Future ：获取异步返回的结果需要使用轮询的方式，消耗cup

        ExecutorService executorService = Executors.newFixedThreadPool(10);

        Future<String> future = executorService.submit(()->{
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "future";
        });
        while(true){
            if(future.isDone()){
                System.out.println(future.get());
                break;
            }
        }

CompletableFuture：采用观察者模式，阻塞获取异步返回的结果，性能得到优化

		 System.out.println("=============CompletableFuture===================");


        CompletableFuture testFuture1 = CompletableFuture.supplyAsync(()->{
            return "丽丽1";
        }).thenApply((element)->{
            System.out.println("testFuture1后续操作："+element);
            return "丽丽2";
        });

        System.out.println(testFuture1.get());


        System.out.println("=============CompletableFuture===================");


        CompletableFuture testFuture2 = CompletableFuture.supplyAsync(()->{
            return "丽丽1";
        }).thenAccept((element)->{
            System.out.println("testFuture2后续操作："+element);
        });

        System.out.println(testFuture2.get());

CompletableFuture的使用明细

官方文档：https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/asyncio

* runAsync 无返回值
* supplyAsync 有返回值
*
* thenAccept 无返回值
* thenApply 有返回值
* thenRun 不关心上一步执行结果，执行下一个操作
* get() 为阻塞获取 可设置超时时间 避免长时间阻塞

实现接口 AsyncFunction 用于请求分发

定义一个callback回调函数，该函数用于取出异步请求的返回结果，并将返回的结果传递给ResultFuture 

对DataStream的数据使用Async操作

例子

/**
 * An implementation of the 'AsyncFunction' that sends requests and sets the callback.
  *  通过向数据库发送异步请求并设置回调方法
 */
class AsyncDatabaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> {

    /** The database specific client that can issue concurrent requests with callbacks 
     可以异步请求的特定数据库的客户端 */
    private transient DatabaseClient client;

    @Override
    public void open(Configuration parameters) throws Exception {
        client = new DatabaseClient(host, post, credentials);
    }

    @Override
    public void close() throws Exception {
        client.close();
    }

    @Override
    public void asyncInvoke(String key, final ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {

        // issue the asynchronous request, receive a future for result
        // 发起一个异步请求，返回结果的 future
        final Future<String> result = client.query(key);

        // set the callback to be executed once the request by the client is complete
        // the callback simply forwards the result to the result future
        // 设置请求完成时的回调.将结果传递给 result future
        CompletableFuture.supplyAsync(new Supplier<String>() {
        
            @Override
            public String get() {
                try {
                    return result.get();
                } catch (InterruptedException | ExecutionException e) {
                    // Normally handled explicitly.
                    return null;
                }
            }
        }).thenAccept( (String dbResult) -> {
            resultFuture.complete(Collections.singleton(new Tuple2<>(key, dbResult)));
        });
    }
}

// create the original stream
// 创建一个原始的流
DataStream<String> stream = ...;

// apply the async I/O transformation
// 添加一个 async I/O ,指定超时时间，和进行中的异步请求的最大数量
DataStream<Tuple2<String, String>> resultStream =
    AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100);

注意事项
- Timeout：定义请求超时时间，异步请求多久没完成会被认为是超时了
- Capacity：定义了同时进行的异步请求的数量，可以限制并发请求数量，不会积压过多的请求
- 超时处理：默认当一个异步 I/O 请求超时时，会引发异常并重新启动作业。如果要处理超时，可以覆盖该AsyncFunction的timeout方法来自定义超时之后的处理方式
- 响应结果的顺序：AsyncDataStream包含两种输出模式,
  - unorderedWait无序：响应结果的顺序与异步请求的顺序不同
  - orderedWait有序：响应结果的顺序与异步请求的顺序相同