线程池阻塞问题

查询字典信息，返回数据: ······

public PageUtils queryPage(Map<String, Object> params) {
	//查询分页数据
    Page page = new Page((Integer) params.get(Constant.PAGE),
            (Integer) params.get(Constant.SIZE));
    IPage<FlowCardInfoDto> iPage = this.baseMapper.getPage(page, params);
    List<FlowCardInfoDto> records = iPage.getRecords();
    if (records.isEmpty()) {
        return new PageUtils(iPage);
    }
	//CountDownLatch  等待所有结果返回才进行下一步  size=10
    CountDownLatch latch = new CountDownLatch(records.size());
    //查询卡商对象关系---这里打印：查询字典信息，返回数据: ······
    Map<String, Object> dealBeanMap = dictUtil.getDealBeanMap("dealer_type");
    List<FlowCardInfoDto> collect = records.stream().peek(item -> {
        //查询实时流量
        String dealerBeanName = (String) dealBeanMap.get(item.getDealerNo());
        if (StringUtils.isNotBlank(dealerBeanName)) {
            Runnable runnable = () -> {
                //有此卡商
                try {
                    //*****省略部分代码*****
                    //这里根据分页的数据： iccId查询三方接口数据
                    trafficMap = strategy.trafficQueryDoPost(trafficMap);
                    //*****省略部分代码*****
                } finally {
                    //CountDownLatch 执行完递减
                    latch.countDown();
                }
            };
            //这里将 runnable 交给 flowCardThreadPoolExecutor 线程池
            flowCardThreadPoolExecutor.execute(runnable);
        } else {
            log.info("卡商<{}>处理接口未配置！！！", dealerBeanName);
            latch.countDown();
        }
    }).collect(Collectors.toList());
    try {
        latch.await();
        log.info("====================================");
    } catch (Exception e) {
        log.error("卡商查询主线程等待异常", e);
    }
    iPage.setRecords(collect);
    return new PageUtils(iPage);
}

/**
 * 根据 iccId查询三方数据
 * @return Map
 */
protected Map<String, Object> trafficQueryDoPost(Map<String, Object> param) {
    log.info("开始查询数据");
    Map<String, Object> cardInfoMap = new HashMap<>();
    cardInfoMap.put("cardNumber", param.get("iccId"));
    CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {
		//查询数据一交给线程池 flowCardThreadPoolExecutor，http请求
    }, flowCardThreadPoolExecutor).whenComplete((object, throwable) -> cardInfoMap.put("cardInfo", object));
	
    CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {
        //查询数据二交给线程池 flowCardThreadPoolExecutor，http请求
    }, flowCardThreadPoolExecutor).whenComplete((object, throwable) -> cardInfoMap.put("dailyFlow", object));

    CompletableFuture<Object> future3 = CompletableFuture.supplyAsync(() -> {
        //查询数据三交给线程池 flowCardThreadPoolExecutor，http请求
    }, flowCardThreadPoolExecutor).whenComplete((object, throwable) -> cardInfoMap.put("monthlyFlow", object));
    //等待所有结果的
    CompletableFuture<Void> all = CompletableFuture.allOf(future1, future2, future3);

    //阻塞，直到所有任务结束。
    log.info("**：等待所有结果返回");
    all.join();
    log.info("**：所有结果全部返回");
    return cardInfoMap;
}

/**
 * 线程池配置
 * @return Map
 */
@Bean("flowCardThreadPoolExecutor")
public ThreadPoolTaskExecutor flowCardThreadPoolTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(100);
    executor.setKeepAliveSeconds(60);
    executor.setThreadNamePrefix(threadNamePrefix);

    executor.setRejectedExecutionHandler((r, executor1) -> {
        if (!executor1.isShutdown()) {
            try {
                executor1.getQueue().put(r);
            } catch (InterruptedException e) {
                log.error("interruptedException:{}", e.toString());
            }
        }
    });
    // 初始化
    executor.initialize();
    return executor;
}

CompletableFuture<Void> all = CompletableFuture.allOf(future1, future2, future3);

//阻塞，直到所有任务结束。
log.info("**：等待所有结果返回");
try {
    all.get(10, TimeUnit.SECONDS);
} catch (Exception e) {
    log.error("**：等待结果返回异常：", e);
}