CompletableFuture

## 什么是CompletableFuture？
CompletableFuture在Java里面被用于异步编程，异步通常意味着非阻塞，可以使得我们的任务单独运行在与主线程分离的其他线程中，并且通过 回调可以在主线程中得到异步任务的执行状态，是否完成，和是否异常等信息。CompletableFuture实现了Future, CompletionStage接口，实现了Future接口就可以兼容现在有线程池框架，而CompletionStage接口才是异步编程的接口抽象，里面定义多种异步方法，通过这两者集合，从而打造出了强大的CompletableFuture类。

--------------------------

## 简单入门
### 线程的通信
```
// 主线程
CompletableFuture<String> completableFuture=new CompletableFuture<String>();

Runnable runnable=new Runnable() {
    @Override
    public void run() {
		try {
			TimeUnit.SECONDS.sleep(3);
			System.out.println(getThreadName()+"执行子线程.....");
			//在子线程中完成主线程completableFuture的完成
			completableFuture.complete("success");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
};
// 子线程
Thread t1=new Thread(runnable);
t1.start();

//主线程阻塞，等待完成后拿到值
String result=completableFuture.get();
System.out.println(getThreadName()+"1x:  "+result);
```

### 无返回值的异步任务
```
CompletableFuture<Void> future=CompletableFuture.runAsync(new Runnable() {
	@Override
	public void run() {
		try {
			System.out.println(getThreadName()+"正在执行一个没有返回值的异步任务。");
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
});
future.get();
System.out.println(getThreadName()+" end");
```

### 有返回值的异步任务
```
CompletableFuture<String> future=CompletableFuture.supplyAsync(new Supplier<String>() {
	@Override
	public String get() {
		try {
			System.out.println(getThreadName()+"正在执行一个有返回值的异步任务。");
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return "OK";
	}
});
String result = future.get();
System.out.println(getThreadName()+" ans = " + result);
```

---------------------------

## 高级进阶

### 带返回值的异步callback调用
`thenApply`返回一个新的 CompletionStage，当这个阶段正常完成时，将这个阶段的结果传给函数作为参数执行。
```
CompletableFuture<String> init=CompletableFuture.supplyAsync(new Supplier<String>() {
	@Override
	public String get() {
		System.out.println(getThreadName() + "supplyAsync");
		return "123";
	}
});

CompletableFuture<Integer> second = init.thenApply(number->{
	System.out.println(getThreadName()+"thenApply1");
	return Integer.parseInt(number);
});

CompletableFuture<Integer> third = second.thenApply(number->{
	System.out.println(getThreadName()+"thenApply2");
	return number*2;
});

System.out.println(getThreadName() + " => " + third.get());
```

### 不带返回值的异步callback调用
`thenAccept`本身不再返回任何值，适合用于多个callback函数的最后一步操作使用
```
CompletableFuture<String> init=CompletableFuture.supplyAsync(new Supplier<String>() {
	@Override
	public String get() {
		System.out.println(getThreadName() + "supplyAsync");
		return "123";
	}
});

CompletableFuture<Integer> second = init.thenApply(number->{
	System.out.println(getThreadName()+"thenApply1");
	return Integer.parseInt(number);
});

CompletableFuture<Integer> end = second.thenAccept(number->{
	System.out.println(getThreadName()+"thenApply2");
	return number*2;
});

end.get()；
System.out.println(getThreadName() + " => END"  );
```

### 不带返回值也不接受参数的尾处理
```
CompletableFuture.supplyAsync(()->{
    ystem.out.println(getThreadName()+"supplyAsync: 一阶段任务");
    return null;
}).thenRun(()->{
    System.out.println(getThreadName()+"thenRun: 收尾任务");
}).get();
```

### 合并两个有依赖关系的任务
```
CompletableFuture<String>  future1=CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public String get() {
        return "1";
    }
});

CompletableFuture<String>nestedResult = future1.thenCompose(value->
    CompletableFuture.supplyAsync(()->{
        return value+"2";
}));

System.out.println(nestedResult.get());
```

### 合并两个没有依赖关系的任务
```
CompletableFuture<Double>  future1 = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public Double get() {
        return 1.5;
    }
});

CompletableFuture<Double>  future2 = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public Double get() {
        return 2.3;
    }
});

CompletableFuture<Double> result=  d1.thenCombine(d2, (number1,number2)->{
    return  number1+number2;
});

System.out.println(result.get());
```

### 多任务都执行完毕[有数据]
```
CompletableFuture<Double> array[]=new CompletableFuture[3];
for ( int i = 0; i < 3; i++) {
    array[i]=CompletableFuture.supplyAsync(new Supplier<Double>() {
        @Override
        public Double get() {
            return Math.random();
        }
    });
 }

//获取结果的方式一
//       CompletableFuture.allOf(array).get();
//        for(CompletableFuture<Double> cf:array){
//            if(cf.get()>0.6){
//                System.out.println(cf.get());
//            }
//        }

//获取结果的方式二，过滤大于指定数字，在收集输出
List<Double> rs= Stream.of(array).map(CompletableFuture::join).filter(number->number>0.6).collect(Collectors.toList());
System.out.println(rs);

```

### 多任务都执行完毕[无数据]

```
List<String> line = new ArrayList();
List<CompletableFuture<>> totalLines = lines.stream()
   .map(productLine -> CompletableFuture.supplyAsync(() -> {
        //do something async
   }))
.collect(Collectors.toList());
        
//线程聚合等待        
CompletableFuture.allOf(totalLines.toArray(new CompletableFuture[0])).join();
```

### 多任务只要完成一个
```
CompletableFuture<Double> array[]=new CompletableFuture[3];
for ( int i = 0; i < 3; i++) {
    array[i]=CompletableFuture.supplyAsync(new Supplier<Double>() {
        @Override
        public Double get() {
            return i;
        }
    });
}

CompletableFuture<Object> result= CompletableFuture.anyOf(array);
System.out.println(result.get());
```

----------------

## 其余说明

### 异常处理
```
CompletableFuture<Double>  future1 = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public Double get() {
        return 1.5;
    }
}).exceptionally(ex->{
    System.out.println(ex.getMessage());
    return "发生 异常"+ex.getMessage();
});

System.out.println(task.get());
```

### 拦截所有正常异常返回值
```
CompletableFuture<String> task = CompletableFuture.supplyAsync(() -> "name")
    .handle((res, ex) -> {
        System.out.println(ex);
        return res;
    }
);

System.out.println(task.get());
```

### 异步执行设置超时时间
```
# 引入netty
private final static Timer WHEEL_TIMER = new HashedWheelTimer();

CompletableFuture<String> task = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public Double get() {
        return 1.5;
    }
});

final Timeout timeout = WHEEL_TIMER.newTimeout(unused -> {
                    if (!task.isDone()) {
                        task.cancel();
						// 取消后操作
                        LOG.warn("task timed out");
                    }
                },
                2000, TimeUnit.MILLISECONDS);
```

### 加入线程池
```
@Configuration
@Slf4j
@EnableAsync
public class ExecutorConfig {
    @Bean
    public Executor asyncExecutor(TaskExecutionProperties properties) {
        log.info("start async executor");
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
//        配置核心线程数
        threadPoolTaskExecutor.setCorePoolSize(properties.getPool().getCoreSize());
//        配置最大线程数
        threadPoolTaskExecutor.setMaxPoolSize(properties.getPool().getMaxSize());
//        配置队列大小
        threadPoolTaskExecutor.setQueueCapacity(ThreadPoolConstant.QUEUE_CAPACITY);
//        配置线程池中线程的名称前缀
        threadPoolTaskExecutor.setThreadNamePrefix(ThreadPoolConstant.THREAD_NAME_PREFIX);
//   HelloWorldServiceImpl     rejection-policy: 当pool已经达到max size时，如何处理新任务：
//        CallerRunsPolicy: 不在新线程中执行任务，而是由调用者所在的线程来执行；
//        AbortPolicy: 拒绝执行新任务，并抛出RejectedExecutionException异常；
//        DiscardPolicy：丢弃当前将要加入队列的任务；
//        DiscardOldestPolicy：丢弃任务队列中最旧的任务；
        threadPoolTaskExecutor.setRejectedExecutionHandler(
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}

作者：arkMon
链接：https://juejin.cn/post/6914962224029106190
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
```

有余留接口，传入即可
```
CompletableFuture.supplyAsync(Supplier<U> supplier, Executor executor);
CompletableFuture.runAsync(Runnable runnable, Executor executor);
```

## 参考来源
https://cloud.tencent.com/developer/article/1366581