探索CompletableFuture：高效异步编程的利器_competablefuture之批量处理

目录

一、CompletableFuture基本功能安利

二、CompletableFuture使用介绍

（一）任务创建使用

1.supplyAsync创建带有返回值的异步任务

2.runAsync创建没有返回值的异步任务

（二）异步回调使用

1.异步回调：thenApply和thenApplyAsync

2.异步回调：thenCompose和thenComposeAsync

3.异步回调：thenAccept和thenAcceptAsync

4.异步回调：thenRun和thenRunAsync

（三）多任务组合使用

1.多任务组合：thenCombine

2.多任务组合：thenAcceptBoth

3.多任务组合：runAfterBoth

4.多任务组合：applyToEither

5.多任务组合：acceptEither

6.多任务组合：runAfterEither

7.多任务组合：allOf

8.多任务组合：anyOf

（四）结果处理使用

1.结果处理：whenComplete

2.结果处理：exceptionally

3.结果处理：handle

（五）方法混合使用举例

1.多个方法组合使用

2.并发处理批量任务

A.模拟规定校验商品敏感词信息归类

B.模拟从一个上传的商品信息中提炼商品内含的文本数据

C.模拟处理一个商品信息中的敏感词命中情况统计,异步并发

D.模拟并发处理批量任务

E.结果展示

参考文章

---

干货分享，感谢您的阅读！

随着现代软件系统的复杂性和用户需求的多样化，异步编程成为了提升系统性能和响应速度的重要手段。在Java领域，CompletableFuture作为Java 8引入的新特性，提供了强大的异步编程能力，极大地简化了多线程和并发任务的处理。本文将探讨CompletableFuture的基本功能和使用方法，介绍如何利用其提升程序的并发性能和代码的可维护性。

其原理相关内容见：CompletableFuture回调机制的设计与实现_张彦峰ZYF的博客-CSDN博客

一、CompletableFuture基本功能安利

CompletableFuture是JDK8中的新特性，主要用于对JDK5中加入的Future的补充，实现了CompletionStage和Future接口。其基本功能主要如下图，作为一项利器在我们的平时开发中频繁使用。

一般使用上我们主要集中在任务创建、异步回调、多任务组合、结果处理和结果获取，本文主要针对各部分主要功能进行基本的安利使用，同时附上一些源码的文章供分析。

二、CompletableFuture使用介绍

假设我们有一个敏感词系统，可以针对各种信息进行基本的敏感词验证，其核心主要包括三部分功能：文本清洗、敏感词验证、敏感词干预生效。

1. 第一步，文本清洗针对验证的文本将直接去除其中的特殊符号、表情包、隐藏符号、中文简体繁体转换等内容，只保留文本中含有的中英文和数字信息。
2. 第二步，清洗后，将清洗文本与实际相关的词库进行验证来查看是否命中相关敏感词，并给出词库命中的敏感词信息。
3. 第三步，如果存在词库命中的敏感词，同时考虑各词是否有加白以及生效规则（只对某区域生效等），从而确定最终的命中信息。

以上是业务基背景，相关可以采用责任链管道模式实现可见责任链模式（以及变种管道模式）的应用案例_张彦峰ZYF的博客-CSDN博客_责任链模式应用场景责任链在实际开发中的应用还是比较多的，特别是在营销订购系统、审核流转换处理、任务流程处理系统等系统中，其实我们在开发中往往主要应用的主要无非是以下三个场景（起码以我的平时开发的角度来看）：一是无需太关心责任链中各处理流的顺序的简单使用；二是需要关注处理顺序，按责任链条延续处理，每个处理节点均可对请求进行节点的处理， 或将其传递给链上的下个处理节点；三是在处理中和纯的责任链模式在链上只会有一个处理器用于处理业务数据存在差异，需要进行管道模式采用多个处理器都会处理业务数据。针对以上场景进行业务举例和代码书写https://blog.csdn.net/xiaofeng10330111/article/details/123956717?spm=1001.2014.3001.5501其中含有基本业务流程图可方便理解。

（一）任务创建使用

1.supplyAsync创建带有返回值的异步任务

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

我们限定创建用于清洗用户文本有返回值的异步任务，采用以下两种方式进行展示其基本创建：

• 使用默认线程池（ForkJoinPool.commonPool()）异步任务创建-有返回值的异步任务supplyAsync
• 自定义线程池异步任务创建-有返回值的异步任务supplyAsync

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 异步任务创建：有返回值的异步任务supplyAsync
     * 使用默认线程池 + 自定义线程池
     */
    @Test
    public void testSupplyAsync() throws ExecutionException, InterruptedException {
        log.info("异步任务创建-有返回值的异步任务supplyAsync:使用默认线程池（ForkJoinPool.commonPool()）");
        CompletableFuture<String> contentCleanTaskByDefault = CompletableFuture.supplyAsync(() -> {
            log.info("异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】");
            ContentCleanResContext contentCleanResContext = sensitivePipelineExecutor.getContentCleanRes(
                    ContentInfoContext.builder()
                            .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .contentAttr(ContentAttr.builder().build()).build());
 
            return contentCleanResContext.getCleanContent();
        });
        log.info("异步任务获取用户文本清洗结果：【{}】", contentCleanTaskByDefault.get());
 
        log.info("异步任务创建-有返回值的异步任务supplyAsync:自定义线程池");
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<String> contentCleanTaskByDefine = CompletableFuture.supplyAsync(() -> {
            log.info("异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】");
            ContentCleanResContext contentCleanResContext = sensitivePipelineExecutor.getContentCleanRes(
                    ContentInfoContext.builder()
                            .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build());
            return contentCleanResContext.getCleanContent();
        }, threadPoolExecutor);
        log.info("异步任务获取用户文本清洗结果：【{}】", contentCleanTaskByDefine.get());
    }
}

测试结果展示:

重点关注打印日志“异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】”以及各自结果“异步任务获取用户文本清洗结果：【中国张彦峰外卖】”，管道相关处理日志暂时忽略。

2023-01-22 10:47:48,349 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 64] - 异步任务创建-有返回值的异步任务supplyAsync:使用默认线程池（ForkJoinPool.commonPool()）
2023-01-22 10:47:48,354 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 66] - 异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】
2023-01-22 10:47:48,356 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, contentAttr=ContentAttr(belong=null, cityCode=null, source=null, sourceId=null, bizType=null))
2023-01-22 10:47:48,398 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 34 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 10:47:48,434 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 34 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 10:47:48,466 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 31 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 10:47:48,877 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中国张彦峰外卖, contentAttr=ContentAttr(belong=null, cityCode=null, source=null, sourceId=null, bizType=null))
2023-01-22 10:47:48,878 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 75] - 异步任务获取用户文本清洗结果：【中国张彦峰外卖】
2023-01-22 10:47:48,878 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 77] - 异步任务创建-有返回值的异步任务supplyAsync:自定义线程池
2023-01-22 10:47:48,879 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 88] - 异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】
2023-01-22 10:47:48,880 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 10:47:48,882 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中国张彦峰外卖, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 10:47:48,882 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 96] - 异步任务获取用户文本清洗结果：【中国张彦峰外卖】
 

2.runAsync创建没有返回值的异步任务

public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor)

我们限定创建用于清洗用户文本无返回值的异步任务，采用以下两种方式进行展示其基本创建：

• 使用默认线程池（ForkJoinPool.commonPool()）异步任务创建-无返回值的异步任务supplyAsync
• 自定义线程池异步任务创建-无返回值的异步任务supplyAsync

因为无返回值，所以一般在处理中，异步任务对用户文本清洗并刷新数据到指定缓存，并"异步通知进行消息转发，触达各消费端进行后续清理

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
    /**
     * 异步任务创建：没有返回值的异步任务runAsync
     * 使用默认线程池 + 自定义线程池
     */
    @Test
    public void testRunAsync() throws ExecutionException, InterruptedException {
        log.info("异步任务创建-有返回值的异步任务runAsync:使用默认线程池（ForkJoinPool.commonPool()）");
        CompletableFuture<Void> contentCleanTaskByDefault = CompletableFuture.runAsync(() -> {
            log.info("异步任务对用户文本清洗并刷新数据");
            log.info("异步通知进行消息转发，触达各消费端进行后续清理");
        });
 
        log.info("异步任务创建-有返回值的异步任务runAsync:自定义线程池");
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<Void> contentCleanTaskByDefine = CompletableFuture.runAsync(() -> {
            log.info("异步任务对用户文本清洗并刷新数据");
            log.info("异步通知进行消息转发，触达各消费端进行后续清理");
        }, threadPoolExecutor);
    }
}

测试结果展示:

2023-01-22 10:56:35,672 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 105] - 异步任务创建-有返回值的异步任务runAsync:使用默认线程池（ForkJoinPool.commonPool()）
2023-01-22 10:56:35,676 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 111] - 异步任务创建-有返回值的异步任务runAsync:自定义线程池
2023-01-22 10:56:35,676 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 107] - 异步任务对用户文本清洗并刷新数据
2023-01-22 10:56:35,677 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 108] - 异步通知进行消息转发，触达各消费端进行后续清理
2023-01-22 10:56:35,679 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 122] - 异步任务对用户文本清洗并刷新数据
2023-01-22 10:56:35,680 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 123] - 异步通知进行消息转发，触达各消费端进行后续清理

重点关注打印日志“异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】”以及各自结果“异步通知进行消息转发，触达各消费端进行后续清理”，管道相关处理日志暂时忽略。

（二）异步回调使用

1.异步回调：thenApply和thenApplyAsync

public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) 
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn,Executor executor)

我们限定两个任务来完成用户文本的清洗和词库命中情况，采用以下两种方式进行展示，注意这两个方法是有返回值的：

• 异步回调-thenApply使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况
• 异步回调-thenApplyAsync使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况，采用自定义线程池

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 异步回调：thenApply和thenApplyAsync
     * thenApply接收一个函数作为参数，使用该函数处理上一个CompletableFuture调用的结果，并返回一个具有处理结果的Future对象。
     * <p>
     * 使用thenApply方法时子任务与父任务使用的是同一个线程
     * thenApplyAsync在子任务中是另起一个线程执行任务,可以自定义线程池
     */
    @Test
    public void testThenApply() throws ExecutionException, InterruptedException {
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
        CompletableFuture<SensitveHitContext> sensitveHitRes1 = contentCleanRes.thenApply((contentCleanResInfo) ->
                sensitivePipelineExecutor.getSensitveHitRes(contentCleanResInfo));
        log.info("异步回调-thenApply使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况");
        log.info("任务一：将用户文本进行清洗，清洗结果展示【{}】", contentCleanRes.get().getCleanContent());
        log.info("任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【{}】",
                sensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<SensitveHitContext> sensitveHitRes2 = CompletableFuture.supplyAsync(() ->
                        sensitivePipelineExecutor.getContentCleanRes(
                                ContentInfoContext.builder()
                                        .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                        .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                        .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()), threadPoolExecutor)
                .thenApplyAsync((contentCleanResInfo) ->
                        sensitivePipelineExecutor.getSensitveHitRes(contentCleanResInfo), threadPoolExecutor);
        log.info("异步回调-thenApplyAsync使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况");
        log.info("任务一：将用户文本进行清洗，清洗结果展示【{}】", contentCleanRes.get().getCleanContent());
        log.info("任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【{}】",
                sensitveHitRes2.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
    }
}

测试结果展示:

重点关注打印日志“任务一：将用户文本进行清洗，清洗结果展示【中国张彦峰外卖腾讯南京酒精】”以及“任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【[张彦峰, 外卖, 腾讯, 酒精, 南京]】”，管道相关处理日志暂时忽略。

2023-01-22 11:02:55,076 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 144] - 异步回调-thenApply使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况
2023-01-22 11:02:55,079 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:02:55,122 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 38 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:02:55,156 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 31 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:02:55,188 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 32 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:02:55,565 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:02:55,566 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:02:55,566 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 145] - 任务一：将用户文本进行清洗，清洗结果展示【中国张彦峰外卖腾讯南京酒精】
2023-01-22 11:02:55,580 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:02:55,581 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 146] - 任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【[张彦峰, 腾讯, 酒精, 南京]】
2023-01-22 11:02:55,583 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:02:55,583 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 166] - 异步回调-thenApplyAsync使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况
2023-01-22 11:02:55,583 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 167] - 任务一：将用户文本进行清洗，清洗结果展示【中国张彦峰外卖腾讯南京酒精】
2023-01-22 11:02:55,585 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:02:55,586 [thread-processor-2] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:02:55,586 [thread-processor-2] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:02:55,587 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 168] - 任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【[张彦峰, 腾讯, 酒精, 南京]】

2.异步回调：thenCompose和thenComposeAsync

public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)
public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn)
public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor)

我们限定两个任务来完成用户文本的清洗和词库命中情况，采用以下两种方式进行展示，注意这两个方法是有返回值的：

• 异步回调-thenCompose使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况
• 异步回调-thenComposeAsync使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 异步回调：thenCompose和thenComposeAsync
     * thenCompose的参数为一个返回CompletableFuture实例的函数，该函数的参数是先前计算步骤的结果。
     * <p>
     * thenApply转换的是泛型中的类型，返回的是同一个CompletableFuture
     * thenCompose将内部的CompletableFuture调用展开来并使用上一个CompletableFutre调用的结果在下一步的CompletableFuture调用中进行运算，是生成一个新的CompletableFuture。
     */
    @Test
    public void testThenCompose() throws ExecutionException, InterruptedException {
        CompletableFuture<SensitveHitContext> sensitveHitRes1 = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build())).thenCompose(new Function<ContentCleanResContext, CompletionStage<SensitveHitContext>>() {
            @Override
            public CompletionStage<SensitveHitContext> apply(ContentCleanResContext contentCleanResInfo) {
                return CompletableFuture.supplyAsync(new Supplier<SensitveHitContext>() {
                    @Override
                    public SensitveHitContext get() {
                        return sensitivePipelineExecutor.getSensitveHitRes(contentCleanResInfo);
                    }
                });
            }
        });
        log.info("异步回调-thenCompose使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况");
        log.info("【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】命中敏感词情况结果展示【{}】",
                sensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
 
        CompletableFuture<SensitveHitContext> sensitveHitRes2 = CompletableFuture.supplyAsync(() ->
                        sensitivePipelineExecutor.getContentCleanRes(
                                ContentInfoContext.builder()
                                        .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                        .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                        .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()))
                .thenComposeAsync(new Function<ContentCleanResContext, CompletionStage<SensitveHitContext>>() {
                    @Override
                    public CompletionStage<SensitveHitContext> apply(ContentCleanResContext contentCleanResInfo) {
                        return CompletableFuture.supplyAsync(new Supplier<SensitveHitContext>() {
                            @Override
                            public SensitveHitContext get() {
                                return sensitivePipelineExecutor.getSensitveHitRes(contentCleanResInfo);
                            }
                        });
                    }
                });
        log.info("异步回调-thenComposeAsync使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况");
        log.info("【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】命中敏感词情况结果展示【{}】",
                sensitveHitRes2.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
    }
}

测试结果展示:

重点关注打印日志“【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】命中敏感词情况结果展示【[张彦峰, 外卖, 腾讯, 酒精, 南京]】”，管道相关处理日志暂时忽略。

2023-01-22 11:06:38,047 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 197] - 异步回调-thenCompose使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况
2023-01-22 11:06:38,050 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:06:38,138 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 68 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:06:38,196 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 54 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:06:38,254 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 56 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:06:38,774 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:06:38,776 [ForkJoinPool.commonPool-worker-2] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:06:38,787 [ForkJoinPool.commonPool-worker-2] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:06:38,789 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 198] - 【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】命中敏感词情况结果展示【[张彦峰, 腾讯, 酒精, 南京]】
2023-01-22 11:06:38,790 [ForkJoinPool.commonPool-worker-2] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:06:38,792 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 218] - 异步回调-thenComposeAsync使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况
2023-01-22 11:06:38,794 [ForkJoinPool.commonPool-worker-2] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:06:38,797 [ForkJoinPool.commonPool-worker-2] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:06:38,798 [ForkJoinPool.commonPool-worker-2] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:06:38,800 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 219] - 【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】命中敏感词情况结果展示【[张彦峰, 腾讯, 酒精, 南京]】

3.异步回调：thenAccept和thenAcceptAsync

public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) 
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)

我们限定两个任务来完成用户文本的清洗和词库命中情况，采用以下两种方式进行展示，注意这两个方法是没有返回值的：

• 异步回调-thenAccept使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况
• 异步回调-thenAcceptAsync使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况，使用自定义线程池

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 异步回调：thenAccept和thenAcceptAsync
     * 函数式接口Consumer，这个接口只有输入，没有返回值。
     * <p>
     * thenAccep方法时子任务与父任务使用的是同一个线程
     * henAccepAsync在子任务中可能是另起一个线程执行任务
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Test
    public void testThenAccept() throws ExecutionException, InterruptedException {
        log.info("异步回调-thenAccept使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况");
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
        log.info("任务一：将用户文本进行清洗，清洗结果展示【{}】", contentCleanRes.get().getCleanContent());
 
        CompletableFuture<Void> sensitveHitRes = contentCleanRes.thenAccept((contentCleanResInfo) -> {
            SensitveHitContext sensitveHitContext = sensitivePipelineExecutor.getSensitveHitRes(contentCleanResInfo);
            log.info("任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【{}】",
                    sensitveHitContext.getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        });
 
        log.info("异步回调-thenAcceptAsync使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况");
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<Void> sensitveHitRes2 = CompletableFuture.supplyAsync(() -> {
                            ContentCleanResContext contentCleanResContext = sensitivePipelineExecutor.getContentCleanRes(
                                    ContentInfoContext.builder()
                                            .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                            .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                            .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build());
                            log.info("任务一：将用户文本进行清洗，清洗结果展示【{}】", contentCleanResContext.getCleanContent());
                            return contentCleanResContext;
 
                        }, threadPoolExecutor)
                .thenAcceptAsync((contentCleanResInfo) -> {
                    SensitveHitContext sensitveHitContext = sensitivePipelineExecutor.getSensitveHitRes(contentCleanResInfo);
                    log.info("任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【{}】",
                            sensitveHitContext.getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
                }, threadPoolExecutor);
    }
}

测试结果展示:

重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“任务一：将用户文本进行清洗，清洗结果展示【中国张彦峰外卖腾讯南京酒精】”和“任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【[张彦峰, 腾讯, 酒精, 南京]】”，管道相关处理日志暂时忽略。

2023-01-22 11:10:42,946 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 235] - 异步回调-thenAccept使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况
2023-01-22 11:10:42,953 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:10:43,001 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 41 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:10:43,038 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 36 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:10:43,073 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 34 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:10:43,500 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:10:43,501 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 242] - 任务一：将用户文本进行清洗，清洗结果展示【中国张彦峰外卖腾讯南京酒精】
2023-01-22 11:10:43,502 [main] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:10:43,515 [main] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:10:43,516 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 246] - 任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【[张彦峰, 腾讯, 酒精, 南京]】
2023-01-22 11:10:43,516 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 250] - 异步回调-thenAcceptAsync使用，任务一将用户文本进行清洗，任务二回调清洗结果查看数据数据命中敏感词情况
2023-01-22 11:10:43,518 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:10:43,519 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:10:43,520 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 266] - 任务一：将用户文本进行清洗，清洗结果展示【中国张彦峰外卖腾讯南京酒精】
2023-01-22 11:10:43,520 [thread-processor-2] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:10:43,521 [thread-processor-2] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 11:10:43,521 [thread-processor-2] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 272] - 任务二：回调清洗结果查看数据数据命中敏感词情况，命中结果展示【[张彦峰, 腾讯, 酒精, 南京]】

4.异步回调：thenRun和thenRunAsync

public CompletableFuture<Void> thenRun(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

我们限定两个任务来完成用户文本的清洗和词库命中情况，采用以下两种方式进行展示，注意这两个方法是没有入参也没有返回值的：

• 异步回调-thenRun使用，任务一将用户文本进行清洗，任务二进行异步通知清洗完成执行后续操作
• 异步回调-thenRunAsync使用，任务一将用户文本进行清洗，任务二进行异步通知清洗完成执行后续操作

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 异步回调：thenRun和thenRunAsync
     * thenRun表示某个任务执行完成后执行的动作，即回调方法，无入参，无返回值。
     * thenRun会在上一阶段 CompletableFuture计算完成的时候执行一个Runnable，而Runnable并不使用该CompletableFuture计算的结果。
     * <p>
     * thenRun方法时子任务与父任务使用的是同一个线程
     * thenRunAsync在子任务中可能是另起一个线程执行任务
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Test
    public void testThenRun() throws ExecutionException, InterruptedException {
        log.info("异步回调-thenRun使用，任务一将用户文本进行清洗，任务二进行异步通知清洗完成执行后续操作");
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
        log.info("任务一：将用户文本进行清洗，清洗结果展示【{}】", contentCleanRes.get().getCleanContent());
        CompletableFuture<Void> notifyRes1 = contentCleanRes.thenRun(() -> {
            log.info("任务二：通知相关消费者告知已清洗完毕，可执行后续操作！");
        });
 
        log.info("异步回调-thenRunAsync使用，任务一将用户文本进行清洗，任务二进行异步通知清洗完成执行后续操作");
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<Void> notifyRes2 = CompletableFuture.supplyAsync(() -> {
                            ContentCleanResContext contentCleanResContext = sensitivePipelineExecutor.getContentCleanRes(
                                    ContentInfoContext.builder()
                                            .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                            .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                            .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build());
                            log.info("任务一：将用户文本进行清洗，清洗结果展示【{}】", contentCleanResContext.getCleanContent());
                            return contentCleanResContext;
 
                        }
                        , threadPoolExecutor)
                .thenRunAsync(() -> {
                    log.info("任务二：通知相关消费者告知已清洗完毕，可执行后续操作！");
                }, threadPoolExecutor);
    }
}

测试结果展示:

重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“任务一：将用户文本进行清洗，清洗结果展示【中国张彦峰外卖腾讯南京酒精】”和“任务二：通知相关消费者告知已清洗完毕，可执行后续操作！”，管道相关处理日志暂时忽略。

2023-01-22 11:15:06,816 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 290] - 异步回调-thenRun使用，任务一将用户文本进行清洗，任务二进行异步通知清洗完成执行后续操作
2023-01-22 11:15:06,829 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:15:06,885 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 50 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:15:06,953 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 66 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:15:07,035 [ForkJoinPool.commonPool-worker-9] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 80 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:15:07,780 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:15:07,780 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 297] - 任务一：将用户文本进行清洗，清洗结果展示【中国张彦峰外卖腾讯南京酒精】
2023-01-22 11:15:07,781 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 299] - 任务二：通知相关消费者告知已清洗完毕，可执行后续操作！
2023-01-22 11:15:07,781 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 302] - 异步回调-thenRunAsync使用，任务一将用户文本进行清洗，任务二进行异步通知清洗完成执行后续操作
2023-01-22 11:15:07,783 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:15:07,786 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:15:07,787 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 318] - 任务一：将用户文本进行清洗，清洗结果展示【中国张彦峰外卖腾讯南京酒精】
2023-01-22 11:15:07,788 [thread-processor-2] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 324] - 任务二：通知相关消费者告知已清洗完毕，可执行后续操作！

（三）多任务组合使用

1.多任务组合：thenCombine

public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)

我们限定用户文本的清洗完成清洗，然后异步查看企业合规管控处理查看数据数据命中敏感词情况和正则校验处理查看数据数据命中敏感词情况，只要有一个词库反馈不命中就认为商品可以售卖，该方法有具体返回值。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 多任务组合：thenCombine
     * 合并两个线程任务的结果，并进一步处理，该方法有返回值。
     */
    @Test
    public void testThenCombine() throws ExecutionException, InterruptedException {
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.COMPLIANCE).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                complianceSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.REGULAR).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                regularSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<String> result = complianceSensitveHitRes.thenCombineAsync(regularSensitveHitRes, (res1, res2) -> {
            if (CollectionUtils.isEmpty(res1.getHitWords()) || CollectionUtils.isEmpty(res2.getHitWords())) {
                /*只要有一个词库反馈不命中就认为商品可以售卖*/
                return "当前商品可以售卖";
            }
            List<SensitiveWord> hitWords = Lists.newArrayList();
            hitWords.addAll(res1.getHitWords());
            hitWords.addAll(res2.getHitWords());
            return "当前商品不可以售卖，商品信息中包含敏感词" + hitWords.stream().map(SensitiveWord::getSensitive).collect(Collectors.toList());
        });
        log.info("组合任务一和二结论：{}", result.get());
    }
}

测试结果展示:

重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】”和“任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]】”，以及整合结论“组合任务一和二结论：当前商品不可以售卖，商品信息中包含敏感词[张彦峰, 酒精, 南京]”，管道相关处理日志暂时忽略。

2023-01-22 11:30:49,847 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:30:49,893 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 45 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:30:49,931 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 37 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:30:49,967 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 35 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:30:50,401 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:30:50,404 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 357] - 任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】
2023-01-22 11:30:50,415 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 366] - 任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]】
2023-01-22 11:30:50,416 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 378] - 组合任务一和二结论：当前商品不可以售卖，商品信息中包含敏感词[张彦峰, 酒精, 南京]

2.多任务组合：thenAcceptBoth

public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor)

我们限定用户文本的清洗完成清洗，然后异步查看企业合规管控处理查看数据数据命中敏感词情况和正则校验处理查看数据数据命中敏感词情况，只要有一个词库反馈不命中就认为商品可以售卖异步处理商品上架，否则对商品直接下架处理并告知下架理由为存在敏感词，该方法没有返回值，相关操作建议进行异步操作处理。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 多任务组合：thenAcceptBoth
     * 无返回值
     * 当两个CompletionStage都正常完成计算的时候，就会执行提供的action消费两个异步的结果
     */
    @Test
    public void testThenAcceptBoth() throws ExecutionException, InterruptedException {
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.COMPLIANCE).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                complianceSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.REGULAR).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                regularSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<Void> result = complianceSensitveHitRes.thenAcceptBoth(regularSensitveHitRes, (res1, res2) -> {
            if (CollectionUtils.isEmpty(res1.getHitWords()) || CollectionUtils.isEmpty(res2.getHitWords())) {
                /*只要有一个词库反馈不命中就认为商品可以售卖*/
                log.info("组合任务一和二结论异步处理商品上架");
                return;
            }
            List<SensitiveWord> hitWords = Lists.newArrayList();
            hitWords.addAll(res1.getHitWords());
            hitWords.addAll(res2.getHitWords());
            log.info("组合任务一和二结论异步处理商品下架，并告知下架理由为存在敏感词【{}】", hitWords.stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        });
    }
}

测试结果展示:

重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】”和“任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]】”，以及整合结论“组合任务一和二结论异步处理商品下架，并告知下架理由为存在敏感词【[张彦峰, 酒精, 南京]】”，管道相关处理日志暂时忽略。

2023-01-22 11:35:13,204 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:35:13,253 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 49 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:35:13,307 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 51 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:35:13,362 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 53 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:35:14,232 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:35:14,236 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 411] - 任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】
2023-01-22 11:35:14,260 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 420] - 任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]】
2023-01-22 11:35:14,261 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 431] - 组合任务一和二结论异步处理商品下架，并告知下架理由为存在敏感词【[张彦峰, 酒精, 南京]】

3.多任务组合：runAfterBoth

public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action)
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action)
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor)

我们限定用户文本的清洗完成清洗，然后异步查看企业合规管控处理查看数据数据命中敏感词情况和正则校验处理查看数据数据命中敏感词情况，只要有一个词库反馈不命中就认为商品可以售卖异步处理商品上架，否则对商品直接下架处理并告知下架理由为存在敏感词，该方法没有入参也没有返回值，相关操作建议进行异步保存和异步操作处理。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 多任务组合：runAfterBoth
     * 没有入参，也没有返回值
     * 两个线程任务相比较，有任何一个执行完成，就进行下一步操作，不关心运行结果。
     */
    @Test
    public void testRunAfterBoth() throws ExecutionException, InterruptedException {
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.COMPLIANCE).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】，结果暂存redis",
                complianceSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.REGULAR).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}】，结果暂存redis",
                regularSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<Void> result = complianceSensitveHitRes.runAfterBoth(regularSensitveHitRes, () -> {
            log.info("组合任务一和二从缓存中获取暂存数据，以下为模拟");
            if (CollectionUtils.isEmpty(SensitveHitContext.builder().build().getHitWords()) ||
                    CollectionUtils.isEmpty(SensitveHitContext.builder().build().getHitWords())) {
                /*只要有一个词库反馈不命中就认为商品可以售卖*/
                log.info("组合任务一和二结论异步处理商品上架");
                return;
            }
            List<SensitiveWord> hitWords = Lists.newArrayList();
            hitWords.addAll(SensitveHitContext.builder().build().getHitWords());
            hitWords.addAll(SensitveHitContext.builder().build().getHitWords());
            log.info("组合任务一和二结论异步处理商品下架，并告知下架理由为存在敏感词【{}】", hitWords.stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        });
    }
}

测试结果展示:

重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰”、“任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】，结果暂存redis”和“任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[]】，结果暂存redis”，以及整合结论“组合任务一和二结论异步处理商品上架”，管道相关处理日志暂时忽略。

2023-01-22 11:41:12,483 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:41:12,520 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 37 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:41:12,551 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 30 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:41:12,584 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 32 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:41:13,210 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中国张彦峰外卖, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:41:13,218 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 465] - 任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】，结果暂存redis
2023-01-22 11:41:13,241 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 474] - 任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[]】，结果暂存redis
2023-01-22 11:41:13,243 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 477] - 组合任务一和二从缓存中获取暂存数据，以下为模拟
2023-01-22 11:41:13,244 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 481] - 组合任务一和二结论异步处理商品上架

4.多任务组合：applyToEither

public <U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)
public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn)
public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor)

我们限定用户文本的清洗完成清洗，然后异步查看企业合规管控处理查看数据数据命中敏感词情况和正则校验处理查看数据数据命中敏感词情况，只要有一个词库反馈不命中就认为商品可以售卖, 该方式下我们只关心最先返回的命中结果就进行后续操作，一般要求业务两边的校验处理结果是保持一致的，本处只做模拟使用，注意该方式有返回值。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 多任务组合：applyToEither
     * 该方法有返回值
     * 两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的转化操作。
     */
    @Test
    public void testApplyToEither() throws ExecutionException, InterruptedException {
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.COMPLIANCE).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                complianceSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.REGULAR).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}",
                regularSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<String> result = complianceSensitveHitRes.applyToEither(regularSensitveHitRes, res -> {
            if (CollectionUtils.isEmpty(res.getHitWords())) {
                /*只要有一个词库反馈不命中就认为商品可以售卖*/
                return "当前商品可以售卖";
            }
            return "当前商品不可以售卖，商品信息中包含敏感词" + res.getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList());
        });
        log.info("组合任务一和二处理结论：{}", result.get());
    }
}

测试结果展示:

重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】”和“任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]”，以及整合结论“组合任务一和二处理结论：当前商品不可以售卖，商品信息中包含敏感词[张彦峰]”，管道相关处理日志暂时忽略。

2023-01-22 11:46:18,735 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:46:18,803 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 68 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:46:18,865 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 58 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:46:18,919 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 52 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:46:19,668 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:46:19,671 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 521] - 任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】
2023-01-22 11:46:19,682 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 530] - 任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]
2023-01-22 11:46:19,682 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 539] - 组合任务一和二处理结论：当前商品不可以售卖，商品信息中包含敏感词[张彦峰]

5.多任务组合：acceptEither

public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)

我们限定用户文本的清洗完成清洗，然后异步查看企业合规管控处理查看数据数据命中敏感词情况和正则校验处理查看数据数据命中敏感词情况，只要有一个词库反馈不命中就认为商品可以售卖, 该方式下我们只关心最先返回的命中结果就进行后续操作，一般要求业务两边的校验处理结果是保持一致的，本处只做模拟使用，注意该方式没有返回值，处理结果需要异步操作。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 多任务组合：acceptEither
     * 将已经完成任务的执行结果作为方法入参，但是无返回值
     * 两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的消费操作。
     */
    @Test
    public void testAcceptEither() throws ExecutionException, InterruptedException {
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.COMPLIANCE).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                complianceSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.REGULAR).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}",
                regularSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<Void> result = complianceSensitveHitRes.acceptEither(regularSensitveHitRes, (res) -> {
            if (CollectionUtils.isEmpty(res.getHitWords())) {
                /*只要有一个词库反馈不命中就认为商品可以售卖*/
                log.info("组合任务一和二结论异步处理商品上架");
                return;
            }
            log.info("组合任务一和二结论异步处理商品下架，并告知下架理由为存在敏感词【{}】", res.getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        });
    }
}

测试结果展示:

重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】”和“任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]”，以及整合结论“组合任务一和二结论异步处理商品下架，并告知下架理由为存在敏感词【[张彦峰]】”，管道相关处理日志暂时忽略。

2023-01-22 11:50:15,466 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:50:15,510 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 44 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:50:15,548 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 37 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:50:15,593 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 44 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:50:15,977 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:50:15,980 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 572] - 任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】
2023-01-22 11:50:15,990 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 581] - 任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]
2023-01-22 11:50:15,991 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 589] - 组合任务一和二结论异步处理商品下架，并告知下架理由为存在敏感词【[张彦峰]】

6.多任务组合：runAfterEither

public CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action)
public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action)
public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor)

我们限定用户文本的清洗完成清洗，然后异步查看企业合规管控处理查看数据数据命中敏感词情况和正则校验处理查看数据数据命中敏感词情况，只要有一个词库反馈不命中就认为商品可以售卖, 该方式下我们只关心最先返回的命中结果就进行后续操作，一般要求业务两边的校验处理结果是保持一致的，本处只做模拟使用，注意该方式没有入参没有返回值，处理结果需要异步操作。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
 
    /**
     * 多任务组合：runAfterEither
     * 没有入参，也没有返回值
     * 两个线程任务相比较，有任何一个执行完成，就进行下一步操作，不关心运行结果。
     */
    @Test
    public void testRunAfterEither() throws ExecutionException, InterruptedException {
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                                .cleanContent("中國㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.COMPLIANCE).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】，将结果存入redis中",
                complianceSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.REGULAR).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}】，将结果存入redis中",
                regularSensitveHitRes.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<Void> future = complianceSensitveHitRes.runAfterEither(regularSensitveHitRes, () -> {
            log.info("组合任务一和二从缓存中获取暂存数据，以下为模拟：直接按指定内容获取当前缓存中的结果");
            if (CollectionUtils.isEmpty(SensitveHitContext.builder().build().getHitWords())) {
                /*只要有一个词库反馈不命中就认为商品可以售卖*/
                log.info("组合任务一和二结论，从缓存中获取结果不存在敏感词直接异步处理商品上架");
                return;
            }
            log.info("组合任务一和二结论异步处理商品下架，并告知下架理由为存在敏感词【{}】", "張彥峰");
        });
    }
}

测试结果展示:

重点关注打印日志“content=中國㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰”、“任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[]】，将结果存入redis中”和“任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[]】，将结果存入redis中”，以及整合结论“组合任务一和二结论，从缓存中获取结果不存在敏感词直接异步处理商品上架”，管道相关处理日志暂时忽略。

2023-01-22 11:54:33,692 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中國㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:54:33,772 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 80 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 11:54:33,820 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 47 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 11:54:33,858 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 37 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 11:54:34,226 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中国外卖, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 11:54:34,228 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 623] - 任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[]】，将结果存入redis中
2023-01-22 11:54:34,237 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 632] - 任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[]】，将结果存入redis中
2023-01-22 11:54:34,237 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 635] - 组合任务一和二从缓存中获取暂存数据，以下为模拟：直接按指定内容获取当前缓存中的结果
2023-01-22 11:54:34,237 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 638] - 组合任务一和二结论，从缓存中获取结果不存在敏感词直接异步处理商品上架

7.多任务组合：allOf

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

我们限定用户文本的清洗完成清洗，然后异步查看词库情况，按一下两种情况进行分析

• 情况一：当给定的多个异步任务都正常完成后，返回一个新的 CompletableFuture，给定 CompletableFuture 的结果不会反映在返回的 CompletableFuture 中，但可以通过单独检查给定任务来获得结果
• 情况二：当任何一个异步任务异常完成，则返回的CompletableFuture 也会异常完成，并且将该异步任务的异常作为其原因
• 情况三：当存在多个异常完成时，则返回排在前面的异步任务的异常信息。

具体代码如下，各自验证结果分开验证：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 多任务组合：allOf
     * <p>
     * 实现多 CompletableFuture 的同时返回
     * CompletableFuture是多个任务都执行完成后才会执行，只有有一个任务执行异常，则返回的CompletableFuture执行get方法时会抛出异常，如果都是正常执行，则get返回null。
     */
    @Test
    public void testAllOf() throws ExecutionException, InterruptedException {
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        log.info("情况一：当给定的多个异步任务都正常完成后，返回一个新的 CompletableFuture，" +
                "给定 CompletableFuture 的结果不会反映在返回的 CompletableFuture 中，但可以通过单独检查给定任务来获得结果");
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes1 = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.COMPLIANCE).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                complianceSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes1 = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.REGULAR).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                regularSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> thesaurusSensitveHitRes1 = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.THESAURUS).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务三：根据相关业务配置进行相关词库校验匹配查看数据数据命中敏感词情况，命中结果展示【{}】",
                thesaurusSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<Void> result1 = CompletableFuture.allOf(complianceSensitveHitRes1, regularSensitveHitRes1, thesaurusSensitveHitRes1);
        log.info("多任务组合：allOf执行完毕，相关词库敏感词命中情况展示：企业合规管控词库【{}】，正则敏感词库【{}】，业务自身词库【{}】",
                complianceSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()),
                regularSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()),
                thesaurusSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
 
        log.info("情况二：当任何一个异步任务异常完成，则返回的CompletableFuture 也会异常完成，并且将该异步任务的异常作为其原因。");
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes2 = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.COMPLIANCE).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                complianceSensitveHitRes2.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes2 = CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("正则词库管控处理查看数据数据命中敏感词情况异常暂停");
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                regularSensitveHitRes2.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> thesaurusSensitveHitRes2 = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.THESAURUS).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务三：根据相关业务配置进行相关词库校验匹配查看数据数据命中敏感词情况，命中结果展示【{}】",
                thesaurusSensitveHitRes2.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<Void> result2 = CompletableFuture.allOf(complianceSensitveHitRes2, regularSensitveHitRes2, thesaurusSensitveHitRes2);
        log.info("多任务组合：allOf执行完毕，相关词库敏感词命中情况展示：企业合规管控词库【{}】，正则敏感词库【{}】，业务自身词库【{}】",
                complianceSensitveHitRes2.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()),
                regularSensitveHitRes2.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()),
                thesaurusSensitveHitRes2.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
 
        log.info("情况三：当存在多个异常完成时，则返回排在前面的异步任务的异常信息。");
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes3 = CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("企业合规管控处理查看数据数据命中敏感词情况异常暂停");
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                complianceSensitveHitRes3.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes3 = CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("正则合规管控处理查看数据数据命中敏感词情况异常暂停");
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                regularSensitveHitRes3.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> thesaurusSensitveHitRes3 = CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("业务自身词库处理查看数据数据命中敏感词情况异常暂停");
        }, threadPoolExecutor);
        log.info("任务三：根据相关业务配置进行相关词库校验匹配查看数据数据命中敏感词情况，命中结果展示【{}】",
                thesaurusSensitveHitRes3.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<Void> result3 = CompletableFuture.allOf(complianceSensitveHitRes3, regularSensitveHitRes3, thesaurusSensitveHitRes3);
        log.info("多任务组合：allOf执行完毕，相关词库敏感词命中情况展示：企业合规管控词库【{}】，正则敏感词库【{}】，业务自身词库【{}】",
                complianceSensitveHitRes3.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()),
                regularSensitveHitRes3.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()),
                thesaurusSensitveHitRes3.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
    }
}

测试结果展示:

情况一验证结果：重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】”、“任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]”和“任务三：根据相关业务配置进行相关词库校验匹配查看数据数据命中敏感词情况，命中结果展示【[腾讯]】”，以及整合结论“多任务组合：allOf执行完毕，相关词库敏感词命中情况展示：企业合规管控词库【[张彦峰]】，正则敏感词库【[酒精, 南京]】，业务自身词库【[腾讯]】”，管道相关处理日志暂时忽略。

2023-01-22 12:04:07,169 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 669] - 情况一：当给定的多个异步任务都正常完成后，返回一个新的 CompletableFuture，给定 CompletableFuture 的结果不会反映在返回的 CompletableFuture 中，但可以通过单独检查给定任务来获得结果
2023-01-22 12:04:07,171 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:04:07,200 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 29 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 12:04:07,228 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 27 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 12:04:07,255 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 26 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 12:04:07,610 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:04:07,613 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 678] - 任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】
2023-01-22 12:04:07,623 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 687] - 任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]】
2023-01-22 12:04:07,624 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 696] - 任务三：根据相关业务配置进行相关词库校验匹配查看数据数据命中敏感词情况，命中结果展示【[腾讯]】
2023-01-22 12:04:07,625 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 699] - 多任务组合：allOf执行完毕，相关词库敏感词命中情况展示：企业合规管控词库【[张彦峰]】，正则敏感词库【[酒精, 南京]】，业务自身词库【[腾讯]】

情况二验证结果：重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】”、“java.util.concurrent.ExecutionException: java.lang.RuntimeException: 正则词库管控处理查看数据数据命中敏感词情况异常暂停”，管道相关处理日志暂时忽略。

2023-01-22 12:05:22,159 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 704] - 情况二：当任何一个异步任务异常完成，则返回的CompletableFuture 也会异常完成，并且将该异步任务的异常作为其原因。
2023-01-22 12:05:22,162 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:05:22,198 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 36 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 12:05:22,232 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 32 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 12:05:22,259 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 26 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 12:05:22,634 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:05:22,637 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 712] - 任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】
 
java.util.concurrent.ExecutionException: java.lang.RuntimeException: 正则词库管控处理查看数据数据命中敏感词情况异常暂停
 
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
	at org.zyf.javabasic.java8.CompletableFutureTest.testAllOf(CompletableFutureTest.java:718)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.springframework.test.context.junit4.statements.RunBeforeTestExecutionCallbacks.evaluate(RunBeforeTestExecutionCallbacks.java:74)
	at org.springframework.test.context.junit4.statements.RunAfterTestExecutionCallbacks.evaluate(RunAfterTestExecutionCallbacks.java:84)
	at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75)
	at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86)
	at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:251)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
	at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
	at com.intellij.rt.junit.IdeaTestRunner$Repeater$1.execute(IdeaTestRunner.java:38)
	at com.intellij.rt.execution.junit.TestsRepeater.repeat(TestsRepeater.java:11)
	at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:35)
	at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
	at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
Caused by: java.lang.RuntimeException: 正则词库管控处理查看数据数据命中敏感词情况异常暂停
	at org.zyf.javabasic.java8.CompletableFutureTest.lambda$testAllOf$44(CompletableFutureTest.java:715)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

情况三验证结果：重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“java.util.concurrent.ExecutionException: java.lang.RuntimeException: 企业合规管控处理查看数据数据命中敏感词情况异常暂停”，管道相关处理日志暂时忽略。

2023-01-22 12:06:23,329 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 734] - 情况三：当存在多个异常完成时，则返回排在前面的异步任务的异常信息。
2023-01-22 12:06:23,332 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
 
java.util.concurrent.ExecutionException: java.lang.RuntimeException: 企业合规管控处理查看数据数据命中敏感词情况异常暂停
 
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
	at org.zyf.javabasic.java8.CompletableFutureTest.testAllOf(CompletableFutureTest.java:739)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.springframework.test.context.junit4.statements.RunBeforeTestExecutionCallbacks.evaluate(RunBeforeTestExecutionCallbacks.java:74)
	at org.springframework.test.context.junit4.statements.RunAfterTestExecutionCallbacks.evaluate(RunAfterTestExecutionCallbacks.java:84)
	at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75)
	at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86)
	at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:251)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
	at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
	at org.junit.runners.Suite.runChild(Suite.java:128)
	at org.junit.runners.Suite.runChild(Suite.java:27)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
	at com.intellij.rt.junit.IdeaTestRunner$Repeater$1.execute(IdeaTestRunner.java:38)
	at com.intellij.rt.execution.junit.TestsRepeater.repeat(TestsRepeater.java:11)
	at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:35)
	at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
	at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
Caused by: java.lang.RuntimeException: 企业合规管控处理查看数据数据命中敏感词情况异常暂停
	at org.zyf.javabasic.java8.CompletableFutureTest.lambda$testAllOf$43(CompletableFutureTest.java:736)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
 

8.多任务组合：anyOf

public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

我们限定用户文本的清洗完成清洗，然后异步查看词库情况。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 多任务组合：anyOf
     * <p>
     * 多个给定的 CompletableFuture，当其中的任何一个完成时，方法返回这个 CompletableFuture
     * CompletableFuture是多个任务只要有一个任务执行完成，则返回的CompletableFuture执行get方法时会抛出异常，如果都是正常执行，则get返回执行完成任务的结果。
     */
    @Test
    public void testAnyOf() throws ExecutionException, InterruptedException {
        CompletableFuture<ContentCleanResContext> contentCleanRes = CompletableFuture.supplyAsync(() ->
                sensitivePipelineExecutor.getContentCleanRes(
                        ContentInfoContext.builder()
                                .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build()));
 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
 
        CompletableFuture<SensitveHitContext> complianceSensitveHitRes1 = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.COMPLIANCE).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                complianceSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> regularSensitveHitRes1 = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.REGULAR).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【{}】",
                regularSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<SensitveHitContext> thesaurusSensitveHitRes1 = CompletableFuture.supplyAsync(() -> {
            try {
                return (SensitveHitContext) PipelineRouteConfig.getInstance(SensitiveCons.Validate.THESAURUS).handle(contentCleanRes.get());
            } catch (Exception e) {
                return SensitveHitContext.builder().hitWords(Lists.newArrayList()).build();
            }
        }, threadPoolExecutor);
        log.info("任务三：根据相关业务配置进行相关词库校验匹配查看数据数据命中敏感词情况，命中结果展示【{}】",
                thesaurusSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
        CompletableFuture<Object> result1 = CompletableFuture.anyOf(complianceSensitveHitRes1, regularSensitveHitRes1, thesaurusSensitveHitRes1);
        log.info("多任务组合：allOf执行完毕，相关词库敏感词命中情况展示：企业合规管控词库【{}】，正则敏感词库【{}】，业务自身词库【{}】，allOf结果【{}】",
                complianceSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()),
                regularSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()),
                thesaurusSensitveHitRes1.get().getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()),
                ((SensitveHitContext) result1.get()).getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()));
    }
}

测试结果展示:

重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”、“任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】”、“任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]”和“任务三：根据相关业务配置进行相关词库校验匹配查看数据数据命中敏感词情况，命中结果展示【[腾讯]】”，以及整合结论“多任务组合：allOf执行完毕，相关词库敏感词命中情况展示：企业合规管控词库【[张彦峰]】，正则敏感词库【[酒精, 南京]】，业务自身词库【[腾讯]】，allOf结果【[张彦峰]】”，管道相关处理日志暂时忽略。

2023-01-22 12:07:14,047 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:07:14,089 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 42 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 12:07:14,126 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 35 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 12:07:14,156 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 29 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 12:07:14,529 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:07:14,532 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 789] - 任务一：企业合规管控处理查看数据数据命中敏感词情况，命中结果展示【[张彦峰]】
2023-01-22 12:07:14,543 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 798] - 任务二：正则校验处理查看数据数据命中敏感词情况，命中结果展示【[酒精, 南京]】
2023-01-22 12:07:14,544 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 807] - 任务三：根据相关业务配置进行相关词库校验匹配查看数据数据命中敏感词情况，命中结果展示【[腾讯]】
2023-01-22 12:07:14,545 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 810] - 多任务组合：allOf执行完毕，相关词库敏感词命中情况展示：企业合规管控词库【[张彦峰]】，正则敏感词库【[酒精, 南京]】，业务自身词库【[腾讯]】，allOf结果【[张彦峰]】

（四）结果处理使用

1.结果处理：whenComplete

public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) 
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor)

我们限定对用户文本清洗后，异步进行通知，对原返回结果不做干预，有异常则中断，注意该方法是没有返回值的。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 结果处理：whenComplete
     * <p>
     * 一般与exceptionally()配合使用，获取前一个异步线程的结果和异常
     * 不论上一个阶段是正常/异常完成(即不会对阶段的原来结果产生影响)；类似于 try..catch..finanlly 中 finally 代码块，无论是否发生异常都将会执行的。
     * 当某个任务执行完成后，会将执行结果或者执行期间抛出的异常传递给回调方法：
     * 如果是正常执行则异常为null，回调方法对应的CompletableFuture的result和该任务一致，
     */
    @Test
    public void testWhenComplete() throws ExecutionException, InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            log.info("异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】");
            ContentCleanResContext contentCleanResContext = sensitivePipelineExecutor.getContentCleanRes(
                    ContentInfoContext.builder()
                            .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build());
            return contentCleanResContext.getCleanContent() + 12/0;
        }, threadPoolExecutor).whenCompleteAsync((res, excption) -> {
            log.info("异步任务成功执行,进行异步通知和保存，结果是：{},异常是：", res, excption);
        }, threadPoolExecutor);
        log.info("异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】清洗结果:【{}】", future.get());
    }
}

测试结果展示:

重点关注打印日志“异步任务成功执行,进行异步通知和保存，结果是：null,异常是：java.lang.ArithmeticException: / by zero”，管道相关处理日志暂时忽略。

2023-01-22 12:13:46,401 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 837] - 异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】
2023-01-22 12:13:46,404 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:13:46,444 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 36 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 12:13:46,485 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 39 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 12:13:46,515 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 30 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 12:13:46,918 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中国张彦峰外卖, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:13:46,921 [thread-processor-2] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 845] - 异步任务成功执行,进行异步通知和保存，结果是：null,异常是：
java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
	at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
	at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ArithmeticException: / by zero
	at org.zyf.javabasic.java8.CompletableFutureTest.lambda$testWhenComplete$50(CompletableFutureTest.java:843)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
	... 3 common frames omitted
 
java.util.concurrent.ExecutionException: java.lang.ArithmeticException: / by zero
 
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
	at org.zyf.javabasic.java8.CompletableFutureTest.testWhenComplete(CompletableFutureTest.java:847)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.springframework.test.context.junit4.statements.RunBeforeTestExecutionCallbacks.evaluate(RunBeforeTestExecutionCallbacks.java:74)
	at org.springframework.test.context.junit4.statements.RunAfterTestExecutionCallbacks.evaluate(RunAfterTestExecutionCallbacks.java:84)
	at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75)
	at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86)
	at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:251)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
	at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
	at com.intellij.rt.junit.IdeaTestRunner$Repeater$1.execute(IdeaTestRunner.java:38)
	at com.intellij.rt.execution.junit.TestsRepeater.repeat(TestsRepeater.java:11)
	at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:35)
	at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
	at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
Caused by: java.lang.ArithmeticException: / by zero
	at org.zyf.javabasic.java8.CompletableFutureTest.lambda$testWhenComplete$50(CompletableFutureTest.java:843)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
 

2.结果处理：exceptionally

public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn)

我们限定对用户文本清洗后，异步进行通知，对原返回结果无异常正常返回，存在异常做干预做默认兜底返回，注意该方法是有返回值。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 结果处理：exceptionally
     * <p>
     * 一般与whenComplete()配合使用，异常捕获范围包含前面的所有异步线程
     * 异常的结果处理（上一个阶段异常完成才会被执行）；
     * 使用方式类似于 try catch中的catch代码块中异常处理；
     * 当某个任务执行异常时将抛出的异常作为参数传递到回调方法中，如果该任务正常执行，exceptionally方法返回的CompletionStage的result就是该任务正常执行的结果。
     */
    @Test
    public void testExceptionally() throws ExecutionException, InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            log.info("异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】");
            ContentCleanResContext contentCleanResContext = sensitivePipelineExecutor.getContentCleanRes(
                    ContentInfoContext.builder()
                            .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build());
            return contentCleanResContext.getCleanContent() + 12/0;
        }, threadPoolExecutor).exceptionally(excption -> {
            /*可以感知异常，同时返回默认数据*/
            log.info("执行发生异常，返回默认数据，并异步通知相关业务方，异常信息为：" ,excption);
            return "";
        });
        log.info("异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】清洗结果:【{}】", future.get());
    }
}

测试结果展示:

重点关注打印日志“执行发生异常，返回默认数据，并异步通知相关业务方，异常信息为：java.lang.ArithmeticException: / by zero”和“异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】清洗结果:【】”，管道相关处理日志暂时忽略。

2023-01-22 12:17:31,305 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 870] - 异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】
2023-01-22 12:17:31,309 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:17:31,353 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 35 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 12:17:31,392 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 38 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 12:17:31,428 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 34 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 12:17:31,939 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中国张彦峰外卖, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:17:31,942 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 879] - 执行发生异常，返回默认数据，并异步通知相关业务方，异常信息为：
java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
	at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
	at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ArithmeticException: / by zero
	at org.zyf.javabasic.java8.CompletableFutureTest.lambda$testExceptionally$52(CompletableFutureTest.java:876)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
	... 3 common frames omitted
2023-01-22 12:17:31,942 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 882] - 异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】清洗结果:【】

3.结果处理：handle

public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

我们限定对用户文本清洗后，异步进行通知，对原返回结果可以做干预，有异常则可以采用默认返回处理，注意该方法是有返回值的，个人建议实际返回结果需要对内容和异常做处理的信息建议使用改方式。

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
    
    /**
     * 结果处理：handle
     * <p>
     * 一般与whenComplete()配合使用，异常捕获范围包含前面的所有异步线程
     * 产出型方法，即可以对正常完成的结果进行转换，也可以对异常完成的进行补偿一个结果，即可以改变阶段的现状。
     * 跟whenComplete基本一致，区别在于handle的回调方法有返回值，
     * 且handle方法返回的CompletableFuture的result是回调方法的执行结果或者回调方法执行期间抛出的异常，与原始CompletableFuture的result无关。
     */
    @Test
    public void testHandle() throws ExecutionException, InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            log.info("异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】");
            ContentCleanResContext contentCleanResContext = sensitivePipelineExecutor.getContentCleanRes(
                    ContentInfoContext.builder()
                            .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰")
                            .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build());
            return contentCleanResContext.getCleanContent();
        }, threadPoolExecutor).handle((res, excption) -> {
            /*异步方法执行完的后续处理*/
            if (Objects.nonNull(excption)) {
                log.info("执行发生异常，返回默认数据，异常信息为：" + excption);
                return "";
            }
            log.info("异步任务成功执行,上一步的结果是：" + res);
            return res + "(用于测试)";
        });
        log.info("异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】清洗结果:【{}】", future.get());
    }
}

测试结果展示:

重点关注打印日志“异步任务成功执行,上一步的结果是：中国张彦峰外卖”和“异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】清洗结果:【中国张彦峰外卖(用于测试)】”，管道相关处理日志暂时忽略，同时异常的处理此处不展示（有兴趣可按照上面的方式增加异常进行查看）。

2023-01-22 12:21:33,625 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 905] - 异步任务获取用户文本清洗结果：用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】
2023-01-22 12:21:33,628 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:21:33,675 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 38 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 12:21:33,707 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 31 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 12:21:33,737 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 29 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 12:21:34,186 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰, cleanContent=中国张彦峰外卖, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:21:34,187 [thread-processor-1] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 918] - 异步任务成功执行,上一步的结果是：中国张彦峰外卖
2023-01-22 12:21:34,187 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 921] - 异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰】清洗结果:【中国张彦峰外卖(用于测试)】

（五）方法混合使用举例

1.多个方法组合使用

回到刚开始的敏感词系统，将核心功能：文本清洗、敏感词验证、敏感词干预生效在同一个服务中连贯处理：

• 第一步，文本清洗针对验证的文本将直接去除其中的特殊符号、表情包、隐藏符号、中文简体繁体转换等内容，只保留文本中含有的中英文和数字信息。
• 第二步，清洗后，将清洗文本与实际相关的词库进行验证来查看是否命中相关敏感词，并给出词库命中的敏感词信息。
• 第三步，如果存在词库命中的敏感词，同时考虑各词是否有加白以及生效规则（只对某区域生效等），从而确定最终的命中信息。

原文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】中对应[中国，张彦峰,外卖，腾讯, 酒精, 南京]都是各个词库要求的敏感词，但是对于[中国，外卖]是针对业务做的加白处理等，所以实际生效敏感词是[张彦峰, 腾讯, 酒精, 南京]

使用举例代码：

/**
 * @author yanfengzhang
 * @description CompletableFuture使用
 * @date 2022/12/29  23:45
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZYFApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Slf4j
public class CompletableFutureTest {
 
    @Autowired
    private SensitivePipelineExecutor sensitivePipelineExecutor;
 
    /**
     * 多个方法组合使用
     * <p>
     * 测试组合敏感词处理逻辑
     */
    @Test
    public void testSensitiveDeal() throws ExecutionException, InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
 
        log.info("开始对用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】进行敏感词分析");
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
                    return sensitivePipelineExecutor.getContentCleanRes(
                            ContentInfoContext.builder()
                                    .content("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                    .cleanContent("中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精")
                                    .contentAttr(ContentAttr.builder().bizType(BizType.E_COMMERCE.getType()).cityCode(110010).build()).build());
                }, threadPoolExecutor)
                .thenComposeAsync(new Function<ContentCleanResContext, CompletionStage<SensitveHitContext>>() {
                    @Override
                    public CompletionStage<SensitveHitContext> apply(ContentCleanResContext contentCleanResInfo) {
                        return CompletableFuture.supplyAsync(new Supplier<SensitveHitContext>() {
                            @Override
                            public SensitveHitContext get() {
                                return sensitivePipelineExecutor.getSensitveHitRes(contentCleanResInfo);
                            }
                        });
                    }
                }, threadPoolExecutor)
                .thenApplyAsync((sensitveHitContext) ->
                        sensitivePipelineExecutor.getSensitveEffectiveRes(sensitveHitContext), threadPoolExecutor)
                .handle((res, excption) -> {
                    /*异步方法执行完的后续处理*/
                    if (Objects.nonNull(excption)) {
                        log.info("执行发生异常，返回默认数据，异常信息为：" + excption);
                        return null;
                    }
                    log.info("异步任务成功执行,上一步的结果是：" + res);
                    return res.getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()).toString();
                });
        log.info("异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】命中敏感词为:【{}】", future.get());
    }
}

测试结果展示:

重点关注打印日志“content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精”，以及对应处理结果“异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】命中敏感词为:【[张彦峰, 腾讯, 酒精, 南京]】”，管道相关处理日志暂时忽略。

2023-01-22 12:24:38,526 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 941] - 开始对用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】进行敏感词分析
2023-01-22 12:24:38,532 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:24:38,576 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 38 ms to scan 1 urls, producing 4 keys and 18 values 
2023-01-22 12:24:38,608 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 31 ms to scan 1 urls, producing 4 keys and 13 values 
2023-01-22 12:24:38,642 [thread-processor-1] INFO  org.reflections.Reflections [Reflections.java : 232] - Reflections took 33 ms to scan 1 urls, producing 3 keys and 9 values 
2023-01-22 12:24:39,048 [thread-processor-1] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗(用户文本)构建上下文】, context=ContentInfoContext(content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5))
2023-01-22 12:24:39,050 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 12:24:39,064 [ForkJoinPool.commonPool-worker-9] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【数据清洗结果构建上下文】, context=ContentCleanResContext(isCleanDone=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), reason=null)
2023-01-22 12:24:39,065 [thread-processor-3] INFO  o.z.j.d.r.p.common.CommonHeadHandler [CommonHeadHandler.java : 24] - 管道开始执行：管道名称为【模型实例(敏感词命中)构建上下文】, context=SensitveHitContext(hasHit=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), hitWords=[SensitiveWord(sensitive=张彦峰, sensitiveId=23, kind=4), SensitiveWord(sensitive=腾讯, sensitiveId=23, kind=2), SensitiveWord(sensitive=酒精, sensitiveId=11, kind=5), SensitiveWord(sensitive=南京, sensitiveId=11, kind=5)])
2023-01-22 12:24:39,068 [thread-processor-3] INFO  o.z.j.d.r.p.common.CommonTailHandler [CommonTailHandler.java : 22] - 管道执行完毕：管道名称为【模型实例(敏感词命中)构建上下文】, context=SensitveHitContext(hasHit=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, contentAttr=ContentAttr(belong=null, cityCode=110010, source=null, sourceId=null, bizType=5), hitWords=[SensitiveWord(sensitive=张彦峰, sensitiveId=23, kind=4), SensitiveWord(sensitive=腾讯, sensitiveId=23, kind=2), SensitiveWord(sensitive=酒精, sensitiveId=11, kind=5), SensitiveWord(sensitive=南京, sensitiveId=11, kind=5)])
2023-01-22 12:24:39,068 [thread-processor-3] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 968] - 异步任务成功执行,上一步的结果是：SensitveEffectiveContext(isHit=true, content=中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精, cleanContent=中国张彦峰外卖腾讯南京酒精, hitWords=[SensitiveWord(sensitive=张彦峰, sensitiveId=23, kind=4), SensitiveWord(sensitive=腾讯, sensitiveId=23, kind=2), SensitiveWord(sensitive=酒精, sensitiveId=11, kind=5), SensitiveWord(sensitive=南京, sensitiveId=11, kind=5)], whitedWords=[], complianceIgnoreWords=[], ruleIgnoreWords=[])
2023-01-22 12:24:39,069 [main] INFO  o.z.j.java8.CompletableFutureTest [CompletableFutureTest.java : 971] - 异步任务获取用户文本【中國張彥峰㎵㎶㎷㎸㎹㎺外賣⏳⌚⏰,腾讯,南京酒精】命中敏感词为:【[张彦峰, 腾讯, 酒精, 南京]】

2.并发处理批量任务

假设线上可以对于运营批量新建的商品数据内容进行敏感词分析，针对批量创建的各个商品针对其相关信息（商品名称/商品描述/分类和分类描述/规格名称/属性和属性值/套餐/商品图片特效/商品普通属性和属性值/商家名称/商家公告等）进行敏感词监控并给出对应的结果报告，显然商品数据居多这样的内容建议采用多线程批发处理，此时异步的优势就体现了，具体实现我们按以下方式进行模拟实现：

A.模拟规定校验商品敏感词信息归类

/**
 * @author yanfengzhang
 * @description 校验商品敏感词信息归类
 * @date 2023/1/4  23:00
 */
public enum SensitiveValidateField {
    NAME(1, "商品名称"),
    DESCRIPTION(2, "商品描述"),
    TAG_NAME(3, "分类和分类描述"),
    SPEC(4, "规格名称"),
    ATTR_NAME(5, "属性和属性值"),
    COMBO_MEAL(6, "套餐"),
    SPECIAL_EFFECT_PIC(7, "商品图片特效"),
    CUSTOM_ATTR_NAME(8, "自定义属性"),
    ORDINARY_ATTR_NAME(9, "商品普通属性和属性值"),
    SP_NAME(10, "标品名称"),
    POI_NAME(11, "商家名称"),
    POI_ANNOUNCEMENT(12, "商家公告");
 
    /**
     * 文本归类（商品名称、商品描述、商家公告、商家名称、经营描述、代言信息等）
     */
    private Integer type;
    /**
     * 文本归类描述
     */
    private String desc;
 
    public Integer getType() {
        return type;
    }
 
    public String getDesc() {
        return desc;
    }
 
    SensitiveValidateField(Integer code) {
        this.type = code;
    }
 
    SensitiveValidateField(Integer code, String desc) {
        this.desc = desc;
        this.type = code;
    }
 
    /**
     * 根据文本归类type获取对应的文本归类信息
     *
     * @param type 文本归类type
     * @return
     */
    public static SensitiveValidateField getEnumById(Integer type) {
        for (SensitiveValidateField sensitiveValidateField : SensitiveValidateField.values()) {
            if (sensitiveValidateField.getType().equals(type)) {
                return sensitiveValidateField;
            }
        }
        return null;
    }
 
    /**
     * 根据文本归类信息描述获取对应的文本归类信息
     *
     * @param desc 文本归类描述
     * @return
     */
    public static SensitiveValidateField getEnumByDesc(String desc) {
        for (SensitiveValidateField sensitiveValidateField : SensitiveValidateField.values()) {
            if (sensitiveValidateField.getDesc().equals(desc)) {
                return sensitiveValidateField;
            }
        }
        return null;
    }
 
    /**
     * 判断文本归类type是否在指定范围
     *
     * @param type 文本归类type
     * @return true-在指定范围
     */
    public static boolean isSensitiveValidateField(Integer type) {
        if (null == type) {
            return false;
        }
        for (SensitiveValidateField tempEnum : SensitiveValidateField.values()) {
            if (tempEnum.getType().equals(type)) {
                return true;
            }
        }
        return false;
    }
}

B.模拟从一个上传的商品信息中提炼商品内含的文本数据

    /**
     * 模拟从一个上传的商品信息中提炼商品内含的文本数据
     *
     * @return
     */
    private List<ContentInfoContext> getContentInfosBySpu() {
        List<ContentInfoContext> contentInfoContexts = Lists.newArrayList();
        List<String> spuInfoList = Lists.newArrayList("張彥峰", "肯德基", "外賣", "腾讯", "禁药", "酒精南京",
                "18252066688", "饿了吗", "廉政勤政", "中国", "巴黎圣母院", "莫沫南路", "辉瑞P药", "捷倍安", "极速达", "老城一埚",
                "星即送", "连花清温", "美乐滋", "山茱萸", "欧美齐", "长江鱼", "人气榜第一", "蟹礼券", "茶ta颜悦色", "可食用黄金",
                "安培开席", "摇钱树", "特丁通", "草甘膦", "叫只鸡", "贱男消食片", "维信识别", "龙闩花甲", "鮑家糕点", "至尊至比萨",
                "胖大哥肉蟹煲", "OBLIGI韩式炸鸡", "星芭芭", "欢乐柠檬", "水多活好", "胸胸烈火", "比基妮", "成人艺术", "仿真手枪", "机关枪");
        for (int i = 0; i < 12; i++) {
            String contentInfo = spuInfoList.get(getRandomNumberInRange(0, spuInfoList.size() - 1)) + spuInfoList.get(getRandomNumberInRange(0, spuInfoList.size() - 1))
                    + spuInfoList.get(getRandomNumberInRange(0, spuInfoList.size() - 1));
            contentInfoContexts.add(ContentInfoContext.builder()
                    .content(contentInfo)
                    .cleanContent(contentInfo)
                    .contentAttr(ContentAttr.builder()
                            .belong(SensitiveValidateField.getEnumById(i + 1))
                            .bizType(BizType.E_COMMERCE.getType())
                            .cityCode(110010).build()).build());
        }
        return contentInfoContexts;
    }
 
 
    private static int getRandomNumberInRange(int min, int max) {
        Random r = new Random();
        return r.ints(min, (max + 1)).limit(1).findFirst().getAsInt();
    }

C.模拟处理一个商品信息中的敏感词命中情况统计,异步并发

    /**
     * 模拟处理一个商品信息中的敏感词命中情况统计
     */
    private String  containsSensitiveWordSpu(SpuSensitiveDealCommand spuSensitiveDealCommand) {
        /*1.模拟从一个上传的商品信息中提炼商品内含的文本数据*/
        List<ContentInfoContext> spuContentInfos = getContentInfosBySpu();
        /*2.对商品各项数据进行校验检查*/
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        int isHit = 1;
        int exceptionHit = 2;
        List<CompletableFuture<Map<Integer, String>>> spuSensitiveDealRes = spuContentInfos.stream().map(contentInfoContext -> {
            return CompletableFuture.supplyAsync(() -> {
                        return sensitivePipelineExecutor.getContentCleanRes(contentInfoContext);
                    }, threadPoolExecutor)
                    .thenComposeAsync(new Function<ContentCleanResContext, CompletionStage<SensitveHitContext>>() {
                        @Override
                        public CompletionStage<SensitveHitContext> apply(ContentCleanResContext contentCleanResInfo) {
                            return CompletableFuture.supplyAsync(new Supplier<SensitveHitContext>() {
                                @Override
                                public SensitveHitContext get() {
                                    return sensitivePipelineExecutor.getSensitveHitRes(contentCleanResInfo);
                                }
                            });
                        }
                    }, threadPoolExecutor)
                    .thenApplyAsync((sensitveHitContext) ->
                            sensitivePipelineExecutor.getSensitveEffectiveRes(sensitveHitContext), threadPoolExecutor)
                    .handle((res, excption) -> {
                        Map<Integer, String> hitdetail = Maps.newHashMap();
                        StringBuilder hitInfoRes = new StringBuilder();
                        hitInfoRes.append("【").append(contentInfoContext.getContentAttr().getBelong().getDesc())
                                .append("】[").append(contentInfoContext.getContent()).append("]");
                        /*异步方法执行完的后续处理*/
                        if (Objects.nonNull(excption)) {
                            log.info("执行发生异常，返回默认数据，异常信息为：" + excption);
                            hitInfoRes.append("校验敏感词异常:").append(excption.getMessage());
                            hitdetail.put(exceptionHit, hitInfoRes.toString());
                            return hitdetail;
                        }
                        if (!res.getIsHit()) {
                            return hitdetail;
                        }
                        log.info("异步任务成功执行,上一步的结果是：" + res);
                        hitInfoRes.append("命中敏感词:").append(res.getHitWords().stream().map(SensitiveWord::getSensitive).collect(Collectors.toList()).toString());
                        hitdetail.put(isHit, hitInfoRes.toString());
                        return hitdetail;
                    });
        }).collect(Collectors.toList());
 
        /*3.等待所有并行线程完成，结束单词单词商品处理*/
        CompletableFuture.allOf(spuSensitiveDealRes.toArray(new CompletableFuture[]{})).join();
        threadPoolExecutor.shutdown();
 
        /*4.整合单商品最终结果*/
        List<String> validHits = Lists.newArrayList();
        List<String> exceptionHits = Lists.newArrayList();
        spuSensitiveDealRes.stream().forEach(hitInfo -> {
            Map<Integer, String> hitdetail;
            try {
                hitdetail = hitInfo.get();
            } catch (Exception e) {
                return;
            }
            if (MapUtils.isEmpty(hitdetail)) {
                return;
            }
            if (StringUtils.isNotBlank(hitdetail.get(isHit))) {
                validHits.add(hitdetail.get(isHit));
            }
            if (StringUtils.isNotBlank(hitdetail.get(exceptionHit))) {
                exceptionHits.add(hitdetail.get(exceptionHit));
            }
        });
        /*4.1 如果没有命中敏感词以及异常情况直接返回*/
        String spuName = spuContentInfos.stream().filter(contentInfoContext -> contentInfoContext.getContentAttr().getBelong() == SensitiveValidateField.NAME)
                .findFirst().get().getContent();
        StringBuilder hitInfoRes = new StringBuilder();
        if (CollectionUtils.isEmpty(validHits) && CollectionUtils.isEmpty(exceptionHits)) {
            hitInfoRes.append("商品【").append(spuName).append("】未命中敏感词");
            return hitInfoRes.toString();
        }
        /*4.2 统计其中有效命中信息*/
        if (CollectionUtils.isNotEmpty(validHits)) {
            hitInfoRes.append("商品【").append(spuName).append("】敏感词命中情况统计：\n");
            validHits.stream().forEach(hitInfo -> {
                hitInfoRes.append(hitInfo).append("\n");
            });
        }
        /*4.3 统计其中异常命中信息*/
        if (CollectionUtils.isNotEmpty(exceptionHits)) {
            hitInfoRes.append("商品【").append(spuName).append("】敏感词异常处理情况统计：\n");
            exceptionHits.stream().forEach(hitInfo -> {
                hitInfoRes.append(hitInfo).append("\n");
            });
        }
        return hitInfoRes.toString();
    }

D.模拟并发处理批量任务

    /**
     * 并发处理批量任务
     * <p>
     * 测试组合敏感词处理逻辑
     */
    @Test
    public void testSensitiveBatchDeal() throws ExecutionException, InterruptedException {
        /*定义单个商品处理任务*/
        class SpuSentiveTask implements Callable<String> {
            private final SpuSensitiveDealCommand command;
 
            public SpuSentiveTask(SpuSensitiveDealCommand command) {
                this.command = command;
            }
 
            @Override
            public String call() throws Exception {
                return containsSensitiveWordSpu(command);
            }
        }
 
        /*模拟批量商品 假设前端传入十个商品，我们将十个商品并发执行*/
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,
                16, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
            private AtomicInteger threadCount = new AtomicInteger(1);
 
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-processor-" + threadCount.getAndIncrement());
            }
        });
        List<SpuSentiveTask> spuSentiveTasks = Lists.newArrayList();
        for (int i = 0; i < 10; i++) {
            spuSentiveTasks.add(new SpuSentiveTask(new SpuSensitiveDealCommand()));
        }
 
 
        List<Future<String>> spuSensitiveBatchRes = threadPoolExecutor.invokeAll(spuSentiveTasks);
        threadPoolExecutor.shutdown();
        if (CollectionUtils.isEmpty(spuSensitiveBatchRes)) {
            log.info("本批次商品数据中不存在敏感词！");
            return;
        }
 
        final int[] i = {0};
        spuSensitiveBatchRes.forEach(spuSensitiveRes -> {
            try {
                log.info("商品编号{}对应敏感词校验情况如下：", i[0]);
                log.info(spuSensitiveRes.get());
                i[0]++;
            } catch (Exception e) {
                log.error("商品编号{}对应敏感词校验情况提取失败", i[0]);
                i[0]++;
            }
        });
    }

E.结果展示

原日志因数据太大不在展示，重点关注日志随机选取如下，管道相关处理日志暂时忽略(因为是模拟很多原始数据是随机产生的名称，可以将本问源码直接运行查看最新的随机数据即可，以下是某次随机验证结果)：

商品编号0对应敏感词校验情况如下：
商品【酒精南京18252066688腾讯】敏感词命中情况统计：
【商品名称】[酒精南京18252066688腾讯]命中敏感词:[腾讯, 酒精, 南京]
【分类和分类描述】[肯德基酒精南京茶ta颜悦色]命中敏感词:[肯德基, 酒精, 南京]
【规格名称】[胖大哥肉蟹煲连花清温酒精南京]命中敏感词:[酒精, 南京]
【属性和属性值】[极速达人气榜第一禁药]命中敏感词:[禁药]
【套餐】[腾讯长江鱼机关枪]命中敏感词:[腾讯]
【标品名称】[酒精南京辉瑞P药安培开席]命中敏感词:[酒精, 南京]
【商家名称】[比基妮特丁通酒精南京]命中敏感词:[酒精, 南京]
商品编号1对应敏感词校验情况如下：
商品【人气榜第一酒精南京成人艺术】敏感词命中情况统计：
【商品名称】[人气榜第一酒精南京成人艺术]命中敏感词:[酒精, 南京]
【分类和分类描述】[捷倍安水多活好腾讯]命中敏感词:[腾讯]
【套餐】[張彥峰贱男消食片仿真手枪]命中敏感词:[张彦峰]
【商家公告】[极速达蟹礼券禁药]命中敏感词:[禁药]
商品编号2对应敏感词校验情况如下：
商品【胸胸烈火茶ta颜悦色饿了吗】敏感词命中情况统计：
【规格名称】[摇钱树肯德基山茱萸]命中敏感词:[肯德基]
【商品普通属性和属性值】[肯德基贱男消食片禁药]命中敏感词:[肯德基, 禁药]
商品编号3对应敏感词校验情况如下：
商品【草甘膦巴黎圣母院维信识别】敏感词命中情况统计：
【分类和分类描述】[禁药美乐滋欢乐柠檬]命中敏感词:[禁药]
【自定义属性】[腾讯龙闩花甲草甘膦]命中敏感词:[腾讯]
【商品普通属性和属性值】[張彥峰长江鱼外賣]命中敏感词:[张彦峰]
商品编号4对应敏感词校验情况如下：
商品【贱男消食片18252066688巴黎圣母院】敏感词命中情况统计：
【分类和分类描述】[中国肯德基茶ta颜悦色]命中敏感词:[肯德基]
【规格名称】[腾讯安培开席贱男消食片]命中敏感词:[腾讯]
【属性和属性值】[禁药美乐滋极速达]命中敏感词:[禁药]
【商家名称】[成人艺术星即送腾讯]命中敏感词:[腾讯]
商品编号5对应敏感词校验情况如下：
商品【机关枪張彥峰人气榜第一】敏感词命中情况统计：
【商品名称】[机关枪張彥峰人气榜第一]命中敏感词:[张彦峰]
【商品描述】[禁药贱男消食片叫只鸡]命中敏感词:[禁药]
【属性和属性值】[中国張彥峰老城一埚]命中敏感词:[张彦峰]
【商品图片特效】[比基妮酒精南京胖大哥肉蟹煲]命中敏感词:[酒精, 南京]
【标品名称】[中国星芭芭腾讯]命中敏感词:[腾讯]

参考文章

1.CompletableFuture 使用介绍_Javadoop

2.CompletableFuture使用详解_头未秃的博客-CSDN博客_completablefuture 使用

3.CompletableFuture基本使用_中国一动的博客-CSDN博客_completablefuture使用

4.CompletableFuture使用详解（全网看这一篇就行）_代码搬运工阿新的博客-CSDN博客

5.CompletableFuture使用详解_sermonlizhi的博客-CSDN博客_completablefuture

6.CompletableFuture用法详解_LallanaLee的博客-CSDN博客_completablefuture用法

7.CompletableFuture原理解析 | 老司机撩Java

8.图解CompletableFuture源码_weixin_38592881的博客-CSDN博客

9.深入解读CompletableFuture源码与原理_CoderBruis的博客-CSDN博客_future.isdown()

10.https://blog.csdn.net/qq_52791485/article/details/126980976

11.CompletableFuture源码解析_pngyul的博客-CSDN博客_tryfire