Future和CompletableFuture区别

• Future ：获取异步返回的结果需要使用轮询的方式，消耗cup
  1.         ExecutorService executorService = Executors.newFixedThreadPool(10);
  3.         Future<String> future = executorService.submit(()->{
  4.             try {
  5.                 Thread.sleep(2000);
  6.             } catch (InterruptedException e) {
  7.                 e.printStackTrace();
  8.             }
  9.             return "future";
  10.         });
  11.         while(true){
  12.             if(future.isDone()){
  13.                 System.out.println(future.get());
  14.                 break;
  15.             }
  16.         }

  复制代码
• CompletableFuture：采用观察者模式，阻塞获取异步返回的结果，性能得到优化
  1.                  System.out.println("=============CompletableFuture===================");
  4.         CompletableFuture testFuture1 = CompletableFuture.supplyAsync(()->{
  5.             return "丽丽1";
  6.         }).thenApply((element)->{
  7.             System.out.println("testFuture1后续操作："+element);
  8.             return "丽丽2";
  9.         });
  11.         System.out.println(testFuture1.get());
  14.         System.out.println("=============CompletableFuture===================");
  17.         CompletableFuture testFuture2 = CompletableFuture.supplyAsync(()->{
  18.             return "丽丽1";
  19.         }).thenAccept((element)->{
  20.             System.out.println("testFuture2后续操作："+element);
  21.         });
  23.         System.out.println(testFuture2.get());

  复制代码
• CompletableFuture的使用明细
  
  
  
  • 官方文档：https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/asyncio
    
  1. * runAsync 无返回值
  2. * supplyAsync 有返回值
  3. *
  4. * thenAccept 无返回值
  5. * thenApply 有返回值
  6. * thenRun 不关心上一步执行结果，执行下一个操作
  7. * get() 为阻塞获取 可设置超时时间 避免长时间阻塞

  复制代码
  1. 实现接口 AsyncFunction 用于请求分发
  3. 定义一个callback回调函数，该函数用于取出异步请求的返回结果，并将返回的结果传递给ResultFuture
  5. 对DataStream的数据使用Async操作

  复制代码
  
  
  • 例子
    1. /**
    2. * An implementation of the 'AsyncFunction' that sends requests and sets the callback.
    3.   *  通过向数据库发送异步请求并设置回调方法
    4. */
    5. class AsyncDatabaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> {
    7.     /** The database specific client that can issue concurrent requests with callbacks
    8.      可以异步请求的特定数据库的客户端 */
    9.     private transient DatabaseClient client;
    11.     @Override
    12.     public void open(Configuration parameters) throws Exception {
    13.         client = new DatabaseClient(host, post, credentials);
    14.     }
    16.     @Override
    17.     public void close() throws Exception {
    18.         client.close();
    19.     }
    21.     @Override
    22.     public void asyncInvoke(String key, final ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
    24.         // issue the asynchronous request, receive a future for result
    25.         // 发起一个异步请求，返回结果的 future
    26.         final Future<String> result = client.query(key);
    28.         // set the callback to be executed once the request by the client is complete
    29.         // the callback simply forwards the result to the result future
    30.         // 设置请求完成时的回调.将结果传递给 result future
    31.         CompletableFuture.supplyAsync(new Supplier<String>() {
    32.         
    33.             @Override
    34.             public String get() {
    35.                 try {
    36.                     return result.get();
    37.                 } catch (InterruptedException | ExecutionException e) {
    38.                     // Normally handled explicitly.
    39.                     return null;
    40.                 }
    41.             }
    42.         }).thenAccept( (String dbResult) -> {
    43.             resultFuture.complete(Collections.singleton(new Tuple2<>(key, dbResult)));
    44.         });
    45.     }
    46. }
    48. // create the original stream
    49. // 创建一个原始的流
    50. DataStream<String> stream = ...;
    52. // apply the async I/O transformation
    53. // 添加一个 async I/O ,指定超时时间，和进行中的异步请求的最大数量
    54. DataStream<Tuple2<String, String>> resultStream =
    55.     AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100);

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

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！