SpringBoot | 使用newWorkStealingPool和CompletableFuture进行并发异步处理

梦里梦外; 2024-03-23 22:17 26阅读 0赞

#### 关注wx： `CodingTechWork` ####

## 需求 ##

一个列表操作需要异步处理每个元素，最终需要将列表各个元素的操作结果统一返回，无需关注该列表中的顺序执行。这个线程池不会保证任务的顺序执行，即为WorkStealing抢占式的工作。

## 开发模板 ##

### 线程池配置 ###

@Configuration
    public class ThreadPoolConfig {
        
    
        @Bean
        public ExecutorService getThreadPool(){
        
        	//工作抢占式线程池
            return Executors.newWorkStealingPool(20);
        }
    }

### 多线程调用 ###

/**
         * 线程service
         */
    	@Autowired
        private ExecutorService executorService;
        /**
         * 异步处理列表
         */
        private Boolean asynDo(List<String> idList) {
        
            List<CompletableFuture<Boolean>> futures = new ArrayList<>();
            //CompletableFuture 循环处理
            idList.forEach(id -> {
        
                CompletableFuture<Boolean> future = CompletableFuture.supplyAsync(() ->
                        this.setId(id), executorService);
                //添加到异步汇总结果中
                futures.add(future);
            });
    		//校验总的异步结果
            if (null != futures && !futures.isEmpty()) {
        
            	//等待所有线程执行完毕
                futures.forEach(CompletableFuture::join);
                //处理转发结果
                for (CompletableFuture<Boolean> completableFuture : futures) {
        
                    try {
        
                        Boolean result = completableFuture.get();
                        if (null == result) {
        
                            log.error("asynDo失败，错误信息：结果为空！");
                            return false;
                        }
                        if (!result) {
        
                            //一个失败，则都失败
                            return false;
                        } else {
        
                        	//一个成功，则继续
                        	log.info("id={}成功", id);
                        }
                    } catch (InterruptedException e) {
        
                        log.error("asynDo失败，错误信息：{}", e.getMessage());
                        Thread.currentThread().interrupt();
                        return false;
                    } catch (ExecutionException e) {
        
                        log.error("asynDo失败，错误信息：{}", e.getMessage());
                        return false;
                    }
                }
            }
            return true;
         }
        /**
         * 单个处理
         */
    	 private Boolean setId(String id) {
        
    	 	//TODO 
    		... ...
    	 }