import com.tplhk.thread.completeableFuture.service.MedalInfo;
import com.tplhk.thread.completeableFuture.service.MedalService;
import com.tplhk.thread.completeableFuture.service.UserInfo;
import com.tplhk.thread.completeableFuture.service.UserInfoService;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import java.time.LocalDateTime;
import java.util.Random;
import java.util.concurrent.*;

/**
 * @ClassName : FutureLesson1
 * @Description : 一个例子走进CompletableFuture
 * @Author : jqxu
 * @Date: 2022年4月27日 9:11
 **/
@Slf4j
@SpringBootTest(classes = {FutureLesson1.class})
public class FutureLesson1 {

 /**
 * 1. 一个入门的例子
 *
 * @throws InterruptedException
 * @throws TimeoutException
 * @throws ExecutionException
 */
 @Test
 public void simpleExample() throws InterruptedException, TimeoutException, ExecutionException {
 UserInfoService userInfoService = new UserInfoService();
 MedalService medalService = new MedalService();
 String userId = "666";
 long startTime = System.currentTimeMillis();

 //调用用户服务获取用户基本信息
 CompletableFuture<UserInfo> completableUserInfoFuture = CompletableFuture.supplyAsync(() -> userInfoService.getUserInfo(userId));

 Thread.sleep(300); //模拟主线程其它操作耗时

 CompletableFuture<MedalInfo> completableMedalInfoFuture = CompletableFuture.supplyAsync(() -> medalService.getMedalInfo(userId));

 //下面是等待2秒,如果没有返回,报 TimeoutException
 UserInfo userInfo = completableUserInfoFuture.get(2, TimeUnit.SECONDS);//获取个人信息结果
 // 下面是一直等待,相当于阻塞
 MedalInfo medalInfo = completableMedalInfoFuture.get();//获取勋章信息结果

 log.info(":::userinfo = {}", userInfo.toString());
 log.info(":::medalInfo = {}", medalInfo.toString());
 System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");

 }

 /**
 * 2. 创建异步任务
 * supplyAsync
 * 执行CompletableFuture任务,支持返回值
 * 使用默认内置线程池ForkJoinPool.commonPool() 执行任务,或者自定义线程执行任务
 * runAsync
 * 执行CompletableFuture任务,没有返回值。
 * <p>
 * get() 方法需要抛出 InterruptedException, ExecutionException 异常,join() 方法不用抛异
 *
 * 如果异步任务抛出异常,是看不到异常的,需要调用 get()/join() 方法获取异常信息 并且中止运行
 * 建议 try...catch...处理异常 或者使用 exceptionally 或者使用 handle() 方法处理异常,。
 *
 * 默认线程池的注意点
 * 默认的线程池,处理的线程个数是电脑CPU核数-1。在大量请求过来的时候,处理逻辑复杂的话,响应会很慢。
 * 一般建议使用自定义线程池,优化线程池配置参数。
 * 拒绝策略建议,最好使用AbortPolicy,然后耗时的异步线程,做好线程池隔离!
 * 线程池隔离实现:自定义多个线程池,设置线程名称,隔离不同业务线程池。
 *
 * @throws InterruptedException
 * @throws TimeoutException
 * @throws ExecutionException 本例返回内容:
 * run,关注公众号:捡田螺的小男孩
 * null
 * supply,关注公众号:捡田螺的小男孩捡田螺的小男孩
 */
 @Test
 public void supplyAsyncAndRunAsync() {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 //可以自定义线程池
 ExecutorService executor = Executors.newCachedThreadPool();
 //runAsync的使用
 CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> {
 try {
 Thread.sleep(1 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "run,关注公众号:捡田螺的小男孩");
 }, executor);

 //supplyAsync的使用
 CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> {
 try {
 Thread.sleep(2 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "supply,关注公众号:捡田螺的小男孩");
 return "捡田螺的小男孩";
 }, executor);
 //runAsync的future没有返回值,输出null, join() 是阻塞的
 log.info("{} : {}", LocalDateTime.now(), runFuture.join());
 //supplyAsync的future,有返回值, join() 是阻塞的
 log.info("{} : {}", LocalDateTime.now(), supplyFuture.join());
 executor.shutdown(); // 线程池需要关闭

 }

 /**
 * 3.1 变换 thenRun/thenRunAsync
 * 做完第一个任务后,再做第二个任务
 * 无输入无输出
 * 这两个区别是 thenRunAsync 可以指定自定义线程池,其它 *Async 结尾的也是相同的逻辑
 *
 * @throws InterruptedException
 * @throws TimeoutException
 * @throws ExecutionException
 */
 @Test
 public void thenRunAndthenRunAsync() {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 ExecutorService executor = Executors.newCachedThreadPool();
 CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(3 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "先执行第一个CompletableFuture方法任务");
 return "捡田螺的小男孩";
 }
 );

 CompletableFuture thenRunFuture = orgFuture.thenRun(() -> {
 try {
 Thread.sleep(2 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "接着执行第二个任务");
 });
 log.info("{} : {}", LocalDateTime.now(), thenRunFuture.join());
 }


 /**
 * 3.2 变换 thenAccept/thenAcceptAsync
 * 做完第一个任务后,再做第二个任务
 * 有输入无输出
 * 这两个区别是 thenAcceptAsync 可以指定自定义线程池,其它 *Async 结尾的也是相同的逻辑
 *
 * @throws InterruptedException
 * @throws TimeoutException
 * @throws ExecutionException
 */
 @Test
 public void thenAcceptAndthenAcceptAsync() {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(2 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "先执行第一个CompletableFuture方法任务");
 return "捡田螺的小男孩";
 }
 );

 CompletableFuture thenAcceptFuture = orgFuture.thenAccept((a) -> {
 log.info("{} : {}", LocalDateTime.now(), a);
 try {
 Thread.sleep(2 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "接着执行第二个任务");
 });
 log.info("{} : {}", LocalDateTime.now(), thenAcceptFuture.join());
 }

 /**
 * 3.3 变换 thenApply/thenApplyAsync
 * 做完第一个任务后,再做第二个任务
 * 有输入有输出
 * 这两个区别是 thenApplyAsync 可以指定自定义线程池,其它 *Async 结尾的也是相同的逻辑
 *
 * @throws InterruptedException
 * @throws TimeoutException
 * @throws ExecutionException
 */
 @Test
 public void thenApplyAndthenApplyAsync() {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(20 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "先执行第一个CompletableFuture方法任务");
 return "捡田螺的小男孩";
 }
 );

 CompletableFuture<String> thenApplyAsyncFuture = orgFuture.thenApplyAsync((a) -> {
 log.info("{} : {}", LocalDateTime.now(), a);
 try {
 Thread.sleep(20 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "接着执行第二个任务");
 return "第二个任务完成";
 });
 log.info("{} : {}", LocalDateTime.now(), thenApplyAsyncFuture.join());
 }


 /**
 * 3.4 变换 thenCompose
 * 返回一个 CompletableFuture 或者其他对象。
 * 有输入有输出
 *
 * @throws InterruptedException
 * @throws TimeoutException
 * @throws ExecutionException
 */
 @Test
 public void thenCompose() {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> first = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(10 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "先执行第一个CompletableFuture方法任务");
 return null;
 }
 );

 CompletableFuture<String> second = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(10 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "执行第二个CompletableFuture方法任务");
 return "捡田螺的小男孩second";
 }
 );

 CompletableFuture<String> third = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(10 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "执行第三个CompletableFuture方法任务");
 return "捡田螺的小男孩third";
 }
 );

 CompletableFuture<String> thenComposeFuture = first.thenCompose((a) -> {
 log.info("{} : {}", LocalDateTime.now(), a);
 try {
 Thread.sleep(10 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "执行thenComposeFuture任务");
 return third;
 });
 log.info("{} : {}", LocalDateTime.now(), thenComposeFuture.join());

 }


 /**
 * 4.1 消费异常 exceptionally
 * 消费异常: 一般配置whenComplete() 使用。异常必须抛出,否则 exceptionally 无效
 * 有输入(异常)有输出
 */
 @Test
 public void exceptionally() {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(2 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 // 异常必须抛出,否则 exceptionally 无效
 int i = 1 / 0;
 log.info("{} : {}", LocalDateTime.now(), "运算异常: 1/0");
 return "捡田螺的小男孩";
 }
 );

 CompletableFuture<String> exceptionallyFuture = orgFuture.exceptionally((a) -> {
 log.info("{} : {}", LocalDateTime.now(), a.getMessage());
 try {
 Thread.sleep(1 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "你的程序异常啦";
 });
 log.info("{} : {}", LocalDateTime.now(), exceptionallyFuture.join());
 }

 /**
 * 4.2 消费结果 whenComplete
 * 消费结果: whenComplete 本身没有返回值,但是 get() 得到的是上一个任务的结果
 * 有输入(异常)无输出
 */
 @Test
 public void whenComplete() throws ExecutionException, InterruptedException {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(2 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "有可能运算异常: 1/0");
 if (new Random().nextInt(10) % 2 == 0) {
 log.info("抛了个运算异常");
 int i = 12 / 0;
 }
 return "捡田螺的小男孩";
 }
 );


 /**
 * 不管 orgFuture 是否运算异常,下面代码一定执行。如果 exceptionally 会执行,那么 whenComplete 和 exceptionally 同时执行
 */
 orgFuture.whenComplete((a, e) -> {
 a = a + "!!!!";
 log.info("whenComplete:::{} : {}", LocalDateTime.now(), a);
 try {
 Thread.sleep(3 * 1000);
 } catch (InterruptedException e1) {
 log.info("{} : {}", LocalDateTime.now(), e1.getMessage());

 }
 });

 /**
 * orgFuture 运算异常,下面代码才会执行。如果 exceptionally 会执行,那么 whenComplete 和 exceptionally 同时执行
 */
 orgFuture.exceptionally((e) -> {
 log.info("exceptionally:::{} : {}", LocalDateTime.now(), e.getMessage());
 try {
 Thread.sleep(1 * 1000);
 } catch (InterruptedException e1) {
 e1.printStackTrace();
 }
 return "你的程序异常啦";
 });

 log.info("主程序:::::{} : {}", LocalDateTime.now(), orgFuture.get());
 }

 /**
 * 4.3 消费 handle
 * handle: 相当于 whenComplete + exceptionally.
 * 如果抛了异常, handle(a, e) 接收的参数为 a = null , e有值 ;
 * 如果正常, handle(a, e) 接收的参数为 a = 有值 , e = null ;
 */
 @Test
 public void handle() throws ExecutionException, InterruptedException {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(10 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "有可能运算异常: 1/0");
 if (new Random().nextInt(10) % 2 == 0) {
 log.info("抛了个运算异常");
 int i = 12 / 0;
 }
 return "捡田螺的小男孩";
 }
 );


 /**
 * 不管 orgFuture 是否运算异常,下面代码一定执行。如果 exceptionally 会执行,那么 whenComplete 和 exceptionally 同时执行
 */
 CompletableFuture<String> handle = orgFuture.handle((a, e) -> {
 a = a + "!!!!";
 log.info("handle:::{} : {}", LocalDateTime.now(), a);
 log.info("handle:::{} : {}", LocalDateTime.now(), e);
 try {
 Thread.sleep(10 * 1000);
 } catch (InterruptedException e1) {
 log.info("{} : {}", LocalDateTime.now(), e1.getMessage());

 }
 return "handle 返回结果啦";
 });

 log.info("主程序:::::{} : {}", LocalDateTime.now(), handle.get());
 }


 /**
 * 5.1 组合 and
 * 触发条件:两个任务执行完
 * runAfterBoth/runAfterBothAsync: 无输入无输出
 * thenAcceptBoth/thenAcceptBothAsync: 有输入无输出
 * thenCombine/thenCombineAsync: 有输入有输出
 */
 @Test
 public void thenCombine() throws ExecutionException, InterruptedException {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> first = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(2 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "first......");
 return "捡田螺的小男孩first";
 }
 ).thenApply((a) -> {
 try {
 Thread.sleep(2 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : a:{}, {}", LocalDateTime.now(), a, "first+20s......");
 return "捡田螺的小男孩first";
 });

 CompletableFuture<String> second = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(3 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "second......");
 return "捡田螺的小男孩second";
 }
 );

 CompletableFuture<String> completableFuture = second.thenCombineAsync(first, (f, s) -> {
 log.info("{} : {}", LocalDateTime.now(), "completableFuture......");
 log.info("f = {} : {}", LocalDateTime.now(), f);
 log.info("s = {} : {}", LocalDateTime.now(), s);
 return "两个异步任务的组合执行完了";
 });

 log.info("主程序:::::{} : {}", LocalDateTime.now(), completableFuture.get());

 }


 /**
 * 5.2 组合 or
 * 触发条件:两个任务完成一个 applyToEither / acceptEither / runAfterEither
 * runAfterEither/runAfterEitherAsync: 无输入无输出
 * acceptEither/acceptEitherAsync: 有输入无输出
 * applyToEither/applyToEitherAsync: 有输入有输出
 */
 @Test
 public void applyToEither() throws ExecutionException, InterruptedException {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> first = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(20 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "first......");
 return "捡田螺的小男孩first";
 }
 );

 CompletableFuture<String> second = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(30 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "second......");
 return "捡田螺的小男孩second";
 }
 );

 CompletableFuture<String> applyToEitherFuture = second.applyToEither(first, (f) -> {
 log.info("{} : {}", LocalDateTime.now(), "applyToEitherFuture......");
 log.info("f = {} : {}", LocalDateTime.now(), f);
 return "其中一个异步任务的组合执行完了";
 });

 log.info("主程序:::::{} : {}", LocalDateTime.now(), applyToEitherFuture.get());
 }

 /**
 * 5.3 组合 allof
 * 触发条件:所有任务全部执行完
 */
 @Test
 public void allof() throws ExecutionException, InterruptedException {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> first = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(10 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "first......");
 return "捡田螺的小男孩first";
 }
 );

 CompletableFuture<String> second = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(15 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "second......");
 return "捡田螺的小男孩second";
 }
 );

 CompletableFuture<String> thirth = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(20 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
// int i = 1/0;
 log.info("{} : {}", LocalDateTime.now(), "thirth......");
 return "捡田螺的小男孩thirth";
 }
 );

 CompletableFuture<String> allofFuture = CompletableFuture.allOf(first, second, thirth).handle((a, e) -> {
 log.info("handle:::{} : {}", LocalDateTime.now(), a);
 log.info("handle:::{} : {}", LocalDateTime.now(), e);
 if (e != null) {
 return "有任务出错";
 }
 return "所有任务全部执行完";
 });

 log.info("主程序:::::{} : {}", LocalDateTime.now(), allofFuture.get());
 }

 /**
 * 5.4 组合 anyOf
 * 触发条件:任何一个任务执行完
 */
 @Test
 public void anyOf() throws ExecutionException, InterruptedException {
 log.info("{} : {}", LocalDateTime.now(), "开始执行");
 CompletableFuture<String> first = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(10 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "first......");
 return "捡田螺的小男孩first";
 }
 );

 CompletableFuture<String> second = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(15 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("{} : {}", LocalDateTime.now(), "second......");
 return "捡田螺的小男孩second";
 }
 );

 CompletableFuture<String> thirth = CompletableFuture.supplyAsync(
 () -> {
 try {
 Thread.sleep(20 * 1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
// int i = 1/0;
 log.info("{} : {}", LocalDateTime.now(), "thirth......");
 return "捡田螺的小男孩thirth";
 }
 );

 CompletableFuture<String> anyOfFuture = CompletableFuture.anyOf(first, second, thirth).handle((a, e) -> {
 log.info("handle:::{} : {}", LocalDateTime.now(), a);
 log.info("handle:::{} : {}", LocalDateTime.now(), e);
 if (e != null) {
 return "有任务出错";
 }
 return "所有任务全部执行完";
 });

 log.info("主程序:::::{} : {}", LocalDateTime.now(), anyOfFuture.get());
 }

}