问题

记录一下生产环境出现的问题。。。

几天生产环境有同事反映分页查询一直在转圈查不出来数据，跟我反馈，我也是很积极的去看有什么问题，我以为就是比较常见的问题吧，当我看的时候觉得很奇怪。

有一个分页的接口其实有很多的日志需要打印，为什么只打印了一点日志就没有后续了，然后前台页面一直在转圈圈等待数据的返回，怎么滴？是不喜欢下面的代码不想执行么，很显然不是。

应该正确打印的日志：

20230228162624.png

而实际生产上面打印到下面这条日志结束了。

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

那是什么情况呢？

首先我们说明一下出现问题的场景，emm其实就是一个分页查询。但是呢，分页的数据需要查询一些其他的数据，组装以后返回给前端页面。

简单点就是这样子：

系统调用简图

1. 前端发起请求查询数据
2. 后端根据查询到的数据，组装后请求三方接口（三个接口）
3. 三方接口返回数据给后端服务
4. 后端服务请求完成后返回给页面

我们现在将后端服务详细的执行过程表述下

代码流转图

1. 查询到分页数据后，循环每条数据，使用多线程进行查询三方接口（多线程交给线程池执行）
2. 每个数据的线程在查询数据时有分了三个线程去查询数据（同样交给多线程），数据的线程等待查询的线程相应结果才能往下执行
3. 查询返回的结果组装后返回

正文

下面看下代码时怎么写的。。。

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;
}

上面的代码展示了分页查询的大体逻辑，每次分页的数据条数默认十条带有 iccId 的数据，每条 iccId 都会开辟一个线程来查流量，而这个线程交给线程池，然后每个线程查询流量时，开辟三个线程去查询流量，同样交给了同一个线程池，流量返回后组装完成，一条 iccId 卡才算执行完成进行 countDown()。

问题就出在了线程池上面，我们可以想一下，有关线程池的线程没有日志时怎么回事，没有执行吗？是的，它就是没有执行。

分页查询的十条数据开辟的线程池交给了线程池，瞬间就占用了仅有的十个核心线程，而这十个核心线程每个都必须等待自己开辟的三个核心线程都有结果后才能释放资源，但是这三个线程都在队列里面无法执行（队列未满时，只有核心线程在工作，此时没有普通线程数），造成了查询流量的三十个线程（三个查询流量的线程 * 十条 iccId ）压根就没有机会执行，而核心线程又在等待它们的结果 all.join(); 一直等待 。

造成的结果就是系统服务中凡是涉及到交给线程池执行的操作都不能正常执行。

改进

• 顺序执行：将查询流量的三个三方接口顺序执行，不依靠多线程和线程池。缺点就是时间长，效率低，页面使用的人可能要骂人，服务间调用可能会超时（既然能顺序执行你猜我为什么要用多线程？）

• 线程隔离：另起一个线程配置，将分页数据的线程依旧交给原来的线程池 flowCardThreadPoolExecutor ，将查询流量的三条线程交给另外一个线程池配置，使得两个线程互不影响，查询流量的线程始终有机会执行，就不会造成 flowCardThreadPoolExecutor 线程池阻塞。

保险起见，将查询流量的多线程操作进行如下改动，设置取值的最大等待时间，超过时间抛弃此次请求，保证数据的线程不受影响。

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

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