Java踩坑记系列之线程池-低调大师

Java踩坑记系列之线程池

2020-11-08 549

线程池大家都很熟悉，无论是平时的业务开发还是框架中间件都会用到，大部分都是基于JDK线程池ThreadPoolExecutor做的封装，比如tomcat的线程池，当然也有单独开发的，但都会牵涉到这几个核心参数的设置：核心线程数，等待队列，最大线程数，拒绝策略等。

先说下我们项目组在使用线程池时踩到的坑：

线程池的参数设置一定要结合具体的业务场景，区分I/O密集和CPU密集，如果是I/O密集型业务，核心线程数，workQueue等待队列，最大线程数等参数设置不合理不仅不能发挥线程池的作用，反而会影响现有业务
等待队列workQueue填满后，新创建的线程会优先处理新请求进来的任务，而不是去处理队列里的任务，队列里的任务只能等核心线程数忙完了才能被执行。有可能造成队列里的任务长时间等待，导致队列积压，尤其是I/O密集场景
如果需要得到线程池里的线程执行结果，使用future的方式，拒绝策略不能使用DiscardPolicy，这种丢弃策略虽然不执行子线程的任务，但是还是会返回future对象(其实在这种情况下我们已经不需要线程池返回的结果了)，然后后续代码即使判断了future!=null也没用，这样的话还是会走到future.get()方法，如果get方法没有设置超时时间会导致一直阻塞下去!

伪代码如下：

// 如果线程池已满，新的请求会直接执行拒绝策略
Future<String> future = executor.submit(() -> {
    // 业务逻辑,比如调用第三方接口等耗时操作放在线程池里执行
    return result;
});

// 主流程调用逻辑
if(future != null) // 如果拒绝策略设置不合理还是会走到下面代码
  future.get(超时时间); // 调用方阻塞等待结果返回，直到超时
复制代码

下面就结合实际业务情况逐一进行分析。

当然这些问题一部分是对线程池理解不够导致的，还有一部分是线程池本身的问题。

一. 背景

公司有个接口部分功能使用了线程池，这个功能不依赖核心接口，但有一定的耗时，所以放在线程池里和主线程并行执行，等线程池里的任务执行完通过future.get的方式获取线程池里的线程执行结果，然后合并到主流程的结果里返回给前端，业务场景很简单，大致流程如下：

初衷也是为了不影响主流程的性能，不增加整体响应时间。

但是之前使用的线程池jdk的newCachedThreadPool，因为sonar扫描提示说有内存溢出的风险(最大线程数是Integer.MAX_VALUE)所以当时改成使用原生的ThreadPoolExecutor，通过指定核心线程数和最大线程数，来解决sonar问题。

但是改过的线程池并不适合我们这种I/O密集型的业务场景(大部分业务都是通过调用接口实现的)，当时设置的核心线程数是cpu核数(线上机器是4核)，等待队列是2048，最大线程数是cpu核数*2，从而引发了一系列问题。。。

二. 排查过程

上线后的现象是使用线程池的接口整体响应时间变长，有的甚至到10秒才返回数据，通过线程dump分析发现有大量的线程都阻塞在future.get方法上，如下：

future.get方法会阻塞当前主流程，在超时时间内等待子线程返回结果，如果超时还没结果则结束等待继续执行后续的代码，超时时间设置的是默认接口超时时间10秒(后面已改为200ms)，至此可以确定接口总耗时是因为流程都卡在了future.get这一步了。

但这不是根本原因，future是线程池返回的，伪代码如下：

Future<String> future = executor.submit(() -> {
    // 业务逻辑,比如调用第三方接口等耗时操作放在线程池里执行
    return result;
});
复制代码

通过上面的代码可知future没有结果的原因是提交到线程池里的任务迟迟没有被执行。

那为什么没有执行呢？继续分析线程池的dump文件发现，线程池里的线程数已达到最大数量，满负荷运行，如图：

SubThread是我们自己定义的线程池里线程的名字，8个线程都是runnable状态，说明等待队列里已经塞满任务了，之前设置的队列长度是2048，也就是说还有2048个任务等待执行，这无疑加剧了整个接口的耗时。

线程池的执行顺序是：核心线程数 -> 等待队列 -> 最大线程数 -> 拒绝策略

如果对线程dump分析不太了解的可以看下之前的一篇文章：Windows环境下如何进行线程dump分析，虽然环境不一样但原理类似。

这里基本确定接口耗时变长的主要原因是线程池设置不合理导致的。

另外还有一些偶发问题，就是线上日志显示虽然线程池执行了，但是线程池里的任务却没有记录运行日志，线程池里的任务是调用另外一个服务的接口，和对方接口负责人确认也确实调用了他们的接口，可我们自己的日志里却没有记录下调用报文，经过进一步查看代码发现当时的线程池拒绝策略也被修改过，并不是默认的抛出异常不执行策略AbortPolicy，而是设置的CallerRunsPolicy策略，即交给调用方执行！

也就是说当线程池达到最大负荷时执行的拒绝策略是让主流程去执行提交到线程池里的任务，这样除了进一步加剧整个接口的耗时外，还会导致主流程被hang死，最关键的是无法确定是在哪一步执行提交到线程池的任务。

分析日志埋点可以推断出调用的时间点应该是已经调用完了记录日志的方法，要返回给前端结果的时才执行线程池里任务，此时记录日志的方法已调用过，不会再去打印日志了，而且子任务返回的结果也无法合并到主流程结果里，因为合并主流程结果和线程池任务返回结果的方法也在之前调用过，不会回过头来再调用了，大致流程如下：

其实这种拒绝策略并不适合我们现在的业务场景，因为线程池里的任务不是核心任务，不应该影响主流程的执行。

三. 改进

调整线程池参数，核心线程数基于线上接口的QPS计算，最大线程数参考线上tomcat的最大线程数配置，能够cover住高峰流量，队列设置的尽量小，避免造成任务挤压。关于线程数如何设置会在后续文章中单独讲解。
扩展线程池，封装原生JDK线程池ThreadPoolExecutor，增加对线程池各项指标的监控，包括线程池运行状态、核心线程数、最大线程数、任务等待数、已完成任务数、线程池异常关闭等信息，便于实时监控和定位问题。
重写线程池拒绝策略，主要也是记录超出线程池负载情况下的各项指标情况，以及调用线程的堆栈信息，便于排查分析，通过抛出异常方式中断执行，避免引用的future不为null的问题。
合理调整future.get超时时间，防止阻塞主线程时间过长。

线程池内部流程：

线程池监控和自定义拒绝策略的代码如下，大家可以结合自己的业务场景拿去使用：

package com.javakk;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.MessageFormat;
import java.util.List;
import java.util.concurrent.*;

/**
 * 自定义线程池<p>
 * 1.监控线程池状态及异常关闭等情况<p>
 * 2.监控线程池运行时的各项指标, 比如:任务等待数、已完成任务数、任务异常信息、核心线程数、最大线程数等<p>
 * author: 老K
 */
public class ThreadPoolExt extends ThreadPoolExecutor{

    private static final Logger log = LoggerFactory.getLogger(ThreadPoolExt.class);

    private TimeUnit timeUnit;

    public ThreadPoolExt(int corePoolSize,
                         int maximumPoolSize,
                         long keepAliveTime,
                         TimeUnit unit,
                         BlockingQueue<Runnable> workQueue,
                         ThreadFactory threadFactory,
                         RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        this.timeUnit = unit;
    }

    @Override
    public void shutdown() {
        // 线程池将要关闭事件,此方法会等待线程池中正在执行的任务和队列中等待的任务执行完毕再关闭
        monitor("ThreadPool will be shutdown:");
        super.shutdown();
    }

    @Override
    public List<Runnable> shutdownNow() {
        // 线程池立即关闭事件,此方法会立即关闭线程池,但是会返回队列中等待的任务
        monitor("ThreadPool going to immediately be shutdown:");
        // 记录被丢弃的任务, 暂时只记录日志, 后续可根据业务场景做进一步处理
        List<Runnable> dropTasks = null;
        try {
            dropTasks = super.shutdownNow();
            log.error(MessageFormat.format("ThreadPool discard task count:{0}", dropTasks.size()));
        } catch (Exception e) {
            log.error("ThreadPool shutdownNow error", e);
        }
        return dropTasks;
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        // 监控线程池运行时的各项指标
        monitor("ThreadPool monitor data:");
    }

    @Override
    protected void afterExecute(Runnable r, Throwable ex) {
        if (ex != null) { // 监控线程池中的线程执行是否异常
            log.error("unknown exception caught in ThreadPool afterExecute:", ex);
        }
    }

    /**
     * 监控线程池运行时的各项指标, 比如:任务等待数、任务异常信息、已完成任务数、核心线程数、最大线程数等<p>
     */
    private void monitor(String title){
        try {
            // 线程池监控信息记录, 这里需要注意写ES的时机,尤其是多个子线程的日志合并到主流程的记录方式
            String threadPoolMonitor = MessageFormat.format(
                    "{0}{1}core pool size:{2}, current pool size:{3}, queue wait size:{4}, active count:{5}, completed task count:{6}, " +
                            "task count:{7}, largest pool size:{8}, max pool size:{9}, keep alive time:{10}, is shutdown:{11}, is terminated:{12}, " +
                            "thread name:{13}{14}",
                    System.lineSeparator(), title, this.getCorePoolSize(), this.getPoolSize(),
                    this.getQueue().size(), this.getActiveCount(), this.getCompletedTaskCount(), this.getTaskCount(), this.getLargestPoolSize(),
                    this.getMaximumPoolSize(), this.getKeepAliveTime(timeUnit != null ? timeUnit : TimeUnit.SECONDS), this.isShutdown(),
                    this.isTerminated(), Thread.currentThread().getName(), System.lineSeparator());
            log.info(threadPoolMonitor);
        } catch (Exception e) {
            log.error("ThreadPool monitor error", e);
        }
    }
}
复制代码

自定义拒绝策略代码：

package com.javakk;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.management.*;
import java.text.MessageFormat;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 自定义线程池拒绝策略:<p>
 * 1.记录线程池的核心线程数,活跃数,已完成数等信息,以及调用线程的堆栈信息,便于排查<p>
 * 2.抛出异常中断执行<p>
 * author: 老K
 */
public class RejectedPolicyWithReport implements RejectedExecutionHandler {

    private static final Logger log = LoggerFactory.getLogger(RejectedPolicyWithReport.class);

    private static volatile long lastPrintTime = 0;

    private static final long TEN_MINUTES_MILLS = 10 * 60 * 1000;

    private static Semaphore guard = new Semaphore(1);
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        try {
            String title = "thread pool execute reject policy!!";
            String msg = MessageFormat.format(
                    "{0}{1}core pool size:{2}, current pool size:{3}, queue wait size:{4}, active count:{5}, completed task count:{6}, " +
                            "task count:{7}, largest pool size:{8}, max pool size:{9}, keep alive time:{10}, is shutdown:{11}, is terminated:{12}, " +
                            "thread name:{13}{14}",
                    System.lineSeparator(), title, e.getCorePoolSize(), e.getPoolSize(), e.getQueue().size(), e.getActiveCount(),
                    e.getCompletedTaskCount(), e.getTaskCount(), e.getLargestPoolSize(), e.getMaximumPoolSize(), e.getKeepAliveTime(TimeUnit.SECONDS),
                    e.isShutdown(), e.isTerminated(), Thread.currentThread().getName(), System.lineSeparator());
            log.info(msg);
 threadDump(); // 记录线程堆栈信息包括锁争用信息
        } catch (Exception ex) {
            log.error("RejectedPolicyWithReport rejectedExecution error", ex);
        }
        throw new RejectedExecutionException("thread pool execute reject policy!!");
    }

    /**
     * 获取线程dump信息<p>
     * 注意: 该方法默认会记录所有线程和锁信息虽然方便debug, 使用时最好加开关和间隔调用, 否则可能会增加latency<p>
     * 1.当前线程的基本信息:id,name,state<p>
     * 2.堆栈信息<p>
     * 3.锁相关信息(可以设置不记录)<p>
     *  默认在log记录<p>
     * @return
     */
    private void threadDump() {
        long now = System.currentTimeMillis();
        // 每隔10分钟dump一次
        if (now - lastPrintTime < TEN_MINUTES_MILLS) { 
            return; 
        } 
        if (!guard.tryAcquire()) { 
            return; 
        } 
        // 异步dump线程池信息 
        ExecutorService pool = Executors.newSingleThreadExecutor(); 
        pool.execute(() -> {
            try {
                ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean();
                StringBuilder sb = new StringBuilder();
                for (ThreadInfo threadInfo : threadMxBean.dumpAllThreads(true, true)) {
                    sb.append(getThreadDumpString(threadInfo));
                }
                log.error("thread dump info:", sb.toString());
            } catch (Exception e) {
                log.error("thread dump error", e);
            } finally {
                guard.release();
            }
            lastPrintTime = System.currentTimeMillis();
        });
        pool.shutdown();
    }

    @SuppressWarnings("all")
    private String getThreadDumpString(ThreadInfo threadInfo) {
        StringBuilder sb = new StringBuilder(""" + threadInfo.getThreadName() + """ +
                " Id=" + threadInfo.getThreadId() + " " +
                threadInfo.getThreadState());
        if (threadInfo.getLockName() != null) {
            sb.append(" on " + threadInfo.getLockName());
        }
        if (threadInfo.getLockOwnerName() != null) {
            sb.append(" owned by "" + threadInfo.getLockOwnerName() +
                    "" Id=" + threadInfo.getLockOwnerId());
        }
        if (threadInfo.isSuspended()) {
            sb.append(" (suspended)");
        }
        if (threadInfo.isInNative()) {
            sb.append(" (in native)");
        }
        sb.append('n');
        int i = 0;

        StackTraceElement[] stackTrace = threadInfo.getStackTrace();
        MonitorInfo[] lockedMonitors = threadInfo.getLockedMonitors();
        for (; i < stackTrace.length && i < 32; i++) {
            StackTraceElement ste = stackTrace[i];
            sb.append("tat " + ste.toString());
            sb.append('n');
            if (i == 0 && threadInfo.getLockInfo() != null) {
                Thread.State ts = threadInfo.getThreadState();
                switch (ts) {
                    case BLOCKED:
                        sb.append("t-  blocked on " + threadInfo.getLockInfo());
                        sb.append('n');
                        break;
                    case WAITING:
                        sb.append("t-  waiting on " + threadInfo.getLockInfo());
                        sb.append('n');
                        break;
                    case TIMED_WAITING:
                        sb.append("t-  waiting on " + threadInfo.getLockInfo());
                        sb.append('n');
                        break;
                    default:
                }
            }

            for (MonitorInfo mi : lockedMonitors) {
                if (mi.getLockedStackDepth() == i) {
                    sb.append("t-  locked " + mi);
                    sb.append('n');
                }
            }
        }
        if (i < stackTrace.length) {
            sb.append("t...");
            sb.append('n');
        }

        LockInfo[] locks = threadInfo.getLockedSynchronizers();
        if (locks.length > 0) {
            sb.append("ntNumber of locked synchronizers = " + locks.length);
            sb.append('n');
            for (LockInfo li : locks) {
                sb.append("t- " + li);
                sb.append('n');
            }
        }
        sb.append('n');
        return sb.toString();
    }
}
复制代码

文章来源：javakk.com/188.html

微信关注我们

原文链接：https://my.oschina.net/u/4630893/blog/4708439

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

低功耗设计基础：深入理解Internal Power

之前的文章我们引入了不少低功耗设计的基本概念，同时也介绍过芯片的总体功耗计算方法，其中就包括internal power的定义。今天想深入挖掘internal power以及它在应用过程中有哪些细节需要大家注意。首先还是回顾一下关于数字芯片的功耗定义：主要由静态功耗(Static Power)和动态功耗(Dynamic Power)组成。二者的主要起因可以参考下图：简而言之，静态功耗是芯片中的部分或全部电路在通电但不工作的状态下的功耗；动态功耗是芯片在正常工作时因信号翻转引起的功耗。而今天我们想要深挖的internal power，其实是动态功耗的一种。在数字电路中，动态功耗一般分为internal power和switching power，在此我们将重点介绍一下internal power。很多人都知道internal power的定义是在cell内部由于充放电而产生的功耗，每个cell的功耗数值都可以在.lib文件中通过特定的查表方式获取。其实贡献internal power的因素应该分为两个因素：一是信号翻转时由于NMOS和PMOS同时导通而产生的短暂但巨大的贯穿电流；而...

2020-11-07

855

1. 问题引入小伙伴们都接触过线程，也都会使用线程，今天我们要讲的是线程安全相关的内容，在这之前我们先来看一个简单的代码案例。代码案例： /** * @url: i-code.online * @author: AnonyStar * @time: 2020/10/14 15:39 */ public class ThreadSafaty { //共享变量 static int count = 0; public static void main(String[] args) { //创建线程 Runnable runnable = () -> { for (int i = 0; i < 5; i++) { count ++; try { Thread.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } } }; for (int i = 0; i < 100; i++) { new Thread(runnable,"Thread-"+i).start(); } try { Thr...

2020-11-09

451

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。