CountDownLatch引发的线上问题

图片为证

图片上为事故代码片段，初看没有任何问题

demo.png

1.代码解读

一句话解读：通过阻塞队列结合多线程实现带超时机制的批量多内容多类型的内容安全审核公共方法。

• 多线程
  • 图[2]标，executor是一个核心线程数量1000线程池，拒绝策略：new ThreadPoolExecutor.CallerRunsPolicy()
• 阻塞队列
  • 图[1]标，CountDownLatch实现的阻塞队列
• 超时机制
  • 图[5]标，await接受动态timeout超时时间
• 多内容/类型安全审查
  • 图[3]标，调用第三方内容安全审核接口（*事故根源）

2.事故发现

• 请求接口超时
  • 新上线业务测试接口时，发现业务接口含内容安全审核时100%超时
• mq消息堆积
  • 检查内容安全审核mq-topic，消息大量堆积
• 线上日志诡异
  • 查看日志发现日志只有【接受请求】，无【完成请求】
  • 跟踪日志链路，定位到三方内容安全审核无结果返回
  • 带有超时机制，却未见内容安全审查之后逻辑的日志（最直接的就是无【完成请求】）
  • 十几分钟后三方内容安全审查方法出打印出一个异常日志
• 问题定位
  • 根据日志现象不难发现超时机制未起来作用（理论上不可能）
  • [三方内容安全审查方法响应时间]-十几分钟（相当于未响应）（*事故根源-未做好http超时控制）
  • 初步猜想，三方服务服务请求堆积或者服务期临死机，导致executor线程池被占满
  • executor线程池被占满后，图[2]标处无法获取线程去执行，导致主线程阻塞，无法执行到图[5]标处，超时机制自然无法生效

3.解决过程

• 1.反馈三方技术-对方大概率时重启了服务，接口恢复正常

• 2.联系公司运维重启服务，触发mq重投递，mq正常消费完毕

• 3.尝试寻找线程执行任务时的超时机制：ThreadPoolTaskExecutor 类源码中 void execute(Runnable task, long startTimeout)；这里不知道什么原因，源码竟然忽悠人，没有使用timeout参数

  
  image.png

• 4.（3）步骤无法实行，后尝试其他方式，这里应为业务特殊：http调用三方。查找项目内RestTemplate，未设置任何超时（默认超时：永久）

  
  image.png

• 5.给项目内RestTemplate设置connectTimeout、socketTimeout、connectionRequestTimeout

• 6.在给RestTemplate设置timeout时，发现项目中RestTemplate http连接池未设置

  • 非连接池：默认使用SimpleClientHttpRequestFactory，简单模式，即对每一次HTTP请求均新建一个新的TCP连接，请求结束后则关闭该连接。

  • 连接池：HttpComponentsClientHttpRequestFactory，连接池模式使用PoolingHttpClientConnectionManager，默认单路由5个池，最大10个池(HttpClientBuild.build()）

    
    image.png

• 7.给RestTemplate设置maxTotal、defaultMaxPerRoute

• 注：文章特定参数不理解的可自行查阅理解，本文意在分析过程和大体解决路径

4.总结

• 在使用线程池时，一定要关注异步任务的响应时间
• 使用阻塞队列时要充分考虑是否会执行到阻塞位置
• 涉及http调用时，特别注意timeout和连接池大小配置
• 源码是个好东西，要注意查阅，检验是否如预期

5.思考

图[3]标处如果不是http任务时，应该如何优雅的解决任务本身超时问题？欢迎留下思路~

6.附-文章提到的你可能想要的测试环节

线程池是否沾满、主线程是否阻塞

@Test
    public void test() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(5);
        taskExecutor.setMaxPoolSize(5);
        taskExecutor.setQueueCapacity(0);
        taskExecutor.setKeepAliveSeconds(60);
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.initialize();

        CountDownLatch cdl = new CountDownLatch(20);
        for (int i = 0; i < 20; i++) {
            System.out.println("==== " + i);
            int finalI = i;
            taskExecutor.execute(() -> {
                System.out.println(Thread.currentThread().getId()+": " + finalI);
                try {
                    Thread.sleep(5*1000*1);
                    // test1(template, 4000);
                    // test1(template1, 6000);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    cdl.countDown();
                }
            });
        }
        try {
            cdl.await(50, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("--------------");
    }

http超时机制、http连接池是否够用

同上，注释代码处 //test1(template, timeout)

http接口任务mock，和请求方法

@RequestMapping("/test/timeout")
    public String testTimeout(String str, Integer timeout) throws Exception {
        Thread.sleep(timeout);
        return str;
    }

public void test1(RestTemplate restTemplate, Integer timeout) {
        try {
            String url = "http://localhost:8089/mytest/test/timeout";
            String params = "str=kingtao&timeout="+timeout;

            HttpHeaders httpHeaders = new HttpHeaders();
            httpHeaders.setContentType(MediaType.APPLICATION_FORM_URLENCODED);

            HttpEntity httpEntity = new HttpEntity(params, httpHeaders);

            String returnData = restTemplate.postForEntity(url, httpEntity, String.class).getBody();
            System.out.println(returnData + "--------------");
        } catch (Exception e) {
            System.out.println("error ======================");
            e.printStackTrace();
        }

    }

http timeout、poolnum设置;两个restTemplate Bean

/** 手动注入restTemplate */
    @Bean
    @Primary
    public RestTemplate restTemplate() {
        HttpComponentsClientHttpRequestFactory var1 = new HttpComponentsClientHttpRequestFactory();
        var1.setConnectTimeout(5000); // 连接超时时间：一般不会触发，除非目标服务有问题
        var1.setReadTimeout(5000); // 连接后，socket超时时间（读取数据超时时间），目标服务响应慢时会触发
        var1.setConnectionRequestTimeout(3500); // 从连接池中获取可用连接超时时间，设置的maxTotal、maxPerRoute无可用连接时会触发

        RestTemplate restTemplate = new RestTemplate();
        restTemplate.setRequestFactory(var1);
        return restTemplate;
    }

    @Bean(name = "restTemplate1")
    public RestTemplate restTemplate1() {
        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
        connectionManager.setMaxTotal(500);
        connectionManager.setDefaultMaxPerRoute(200);

        HttpClient httpClient = HttpClientBuilder.create().setConnectionManager(connectionManager).build();

        HttpComponentsClientHttpRequestFactory var1 = new HttpComponentsClientHttpRequestFactory();
        var1.setConnectTimeout(3000);
        var1.setReadTimeout(5000);
        var1.setConnectionRequestTimeout(3500);
        var1.setHttpClient(httpClient);

        RestTemplate restTemplate = new RestTemplate();
        restTemplate.setRequestFactory(var1);
        return restTemplate;
    }