美团动态线程池实践思路开源项目（DynamicTp），线程池源码解析及通知告警篇

V2EX = way to explore

V2EX 是一个关于分享和探索的地方

现在注册

已注册用户请登录

这是一个创建于 957 天前的主题，其中的信息可能已经有所发展或是发生改变。

昨天发布了一篇文章（ https://www.v2ex.com/t/848357#reply5 ），大佬们都好有热情啊，我直呼 V 站牛逼，今天再来一篇，这篇文章我们来聊下动态线程池开源项目（ DynamicTp ）的通知告警模块。目前项目提供以下通知告警功能，每一个通知项都可以独立配置是否开启、告警阈值、告警间隔时间、平台等，具体代码请看 core 模块 notify 包。

1.核心参数变更通知

2.线程池活跃度告警

3.队列容量告警

4.拒绝策略告警

5.任务执行超时告警

6.任务排队超时告警

DynamicTp 项目地址

目前 700star ，感谢你的 star ，欢迎 pr ，业务之余一起给开源贡献一份力量

gitee 地址：https://gitee.com/yanhom/dynamic-tp

github 地址：https://github.com/lyh200/dynamic-tp

系列文章

美团动态线程池实践思路，开源了

动态线程池框架（ DynamicTp ），监控及源码解析篇

动态线程池（ DynamicTp ），动态调整 Tomcat 、Jetty 、Undertow 线程池参数篇

线程池解读

上篇文章里大概讲到了 JUC 线程池的执行流程，我们这里再仔细回顾下，上图是 JUC 下线程池 ThreadPoolExecutor 类的继承体系。

顶级接口 Executor 提供了一种方式，解耦任务的提交和执行，只定义了一个 execute(Runnable command)方法用来提交任务，至于具体任务怎么执行则交给他的实现者去自定义实现。

ExecutorService 接口继承 Executor ，且扩展了生命周期管理的方法、返回 Futrue 的方法、批量提交任务的方法

void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit) throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;

AbstractExecutorService 抽象类继承 ExecutorService 接口，对 ExecutorService 相关方法提供了默认实现，用 RunnableFuture 的实现类 FutureTask 包装 Runnable 任务，交给 execute()方法执行，然后可以从该 FutureTask 阻塞获取执行结果，并且对批量任务的提交做了编排

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}
    
public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

ThreadPoolExecutor 继承 AbstractExecutorService ，采用池化思想管理一定数量的线程来调度执行提交的任务，且定义了一套线程池的生命周期状态，用一个 ctl 变量来同时保存当前池状态（高 3 位）和当前池线程数（低 29 位）。看过源码的小伙伴会发现，ThreadPoolExecutor 类里的方法大量有同时需要获取或更新池状态和池当前线程数的场景，放一个原子变量里，可以很好的保证数据的一致性以及代码的简洁性。

  // 用此变量保存当前池状态（高 3 位）和当前线程数（低 29 位）
  private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); 
  private static final int COUNT_BITS = Integer.SIZE - 3;
  private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

  // runState is stored in the high-order bits
  // 可以接受新任务提交，也会处理任务队列中的任务
  // 结果：111 00000000000000000000000000000
  private static final int RUNNING    = -1 << COUNT_BITS;
  
  // 不接受新任务提交，但会处理任务队列中的任务
  // 结果：000 00000000000000000000000000000
  private static final int SHUTDOWN   =  0 << COUNT_BITS;
  
  // 不接受新任务，不执行队列中的任务，且会中断正在执行的任务
  // 结果：001 00000000000000000000000000000
  private static final int STOP       =  1 << COUNT_BITS;
  
  // 任务队列为空，workerCount = 0 ，线程池的状态在转换为 TIDYING 状态时，会执行钩子方法 terminated()
  // 结果：010 00000000000000000000000000000
  private static final int TIDYING    =  2 << COUNT_BITS;
  
  // 调用 terminated()钩子方法后进入 TERMINATED 状态
  // 结果：010 00000000000000000000000000000
  private static final int TERMINATED =  3 << COUNT_BITS;

  // Packing and unpacking ctl
  // 低 29 位变为 0 ，得到了线程池的状态
  private static int runStateOf(int c)     { return c & ~CAPACITY; }
  // 高 3 位变为为 0 ，得到了线程池中的线程数
  private static int workerCountOf(int c)  { return c & CAPACITY; }
  private static int ctlOf(int rs, int wc) { return rs | wc; }

核心入口 execute()方法执行逻辑如下：

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

可以总结出如下主要执行流程，当然看上述代码会有一些异常分支判断，可以自己顺理加到下述执行主流程里

1.判断线程池的状态，如果不是 RUNNING 状态，直接执行拒绝策略

2.如果当前线程数 < 核心线程池，则新建一个线程来处理提交的任务

3.如果当前线程数 > 核心线程数且任务队列没满，则将任务放入任务队列等待执行

4.如果核心线程池 < 当前线程池数 < 最大线程数，且任务队列已满，则创建新的线程执行提交的任务

5.如果当前线程数 > 最大线程数，且队列已满，则拒绝该任务

addWorker()方法逻辑

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            // 获取当前池状态
            int rs = runStateOf(c);
          
            // (rs >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty()))
            // 1.判断如果线程池状态 > SHUTDOWN ，直接返回 false ，否则 2
            // 2.如果线程池状态 = SHUTDOWN ，并且 firstTask 不为 null 则直接返回 false ，因为 SHUTDOWN 状态的线程池不能在接受新任务，否则 3
            // 3.如果线程池状态 = SHUTDOWN ，并且 firstTask == null ，此时如果任务队列为空，则直接返回 false
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
                int wc = workerCountOf(c);
                // 1.如果当前线程池线程数大于等于 CAPACITY （理论上的最大值 5 亿），则返回 fasle
                // 2.如果创建核心线程情况下当前池线程数 >= corePoolSize ，则返回 false
                // 3.如果创建非核心线程情况下当前池线程数 >= maximumPoolSize ，则返回 false
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // cas 增加当前池线程数量，成功则退出循环    
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                // cas 增加当前池线程数量失败（多线程并发），则重新获取 ctl ，计算出当前线程池状态，如果不等于上述计算的状态 rs ，则说明线程池状态发生了改变，需要跳到外层循环重新进行状态判断，否则执行内部循环
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            // 至此说明线程池状态校验通过，且增加池线程数量成功，则创建一个 Worker 线程来执行任务
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                // 访问 worker set 时需要获取 mainLock 全局锁
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());

                    // 1.当前池状态 < SHUTDOWN ，也就是 RUNNING 状态，如果已经 started ，抛出异常
                    // 2.当前池状态 = SHUTDOWN ，且 firstTask == null ，需要处理任务队列中的任务，如果已经 started ，抛出异常
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        // 刚创建线程添加到 workers 集合中
                        workers.add(w);
                        int s = workers.size();
                        // 判断更新历史最大线程数量
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    // 启动新建线程
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                // 启动失败，workerCount--，workers 里移除该 worker
                addWorkerFailed(w);
        }
        return workerStarted;
    }

线程池中的线程并不是直接用的 Thread 类，而是定义了一个内部工作线程 Worker 类，实现了 AQS 以及 Runnable 接口，然后持有一个 Thread 类的引用及一个 firstTask （创建后第一个要执行的任务），每个 Worker 线程启动后会执行 run()方法，该方法会调用执行外层 runWorker(Worker w)方法

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 1.如果 task 不为空，则作为该线程的第一个任务直接执行
        // 2.如果 task 为空，则通过 getTask()方法从任务队列中获取任务执行
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            // 线程池状态 >= STOP ，则中断线程
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                // 实际执行任务前调用的钩子方法
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 实际执行任务
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    // 任务执行后调用的钩子方法
                    afterExecute(task, thrown);
                }
            } finally {
                // 任务置为 null ，重新获取新任务，完成数++
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 无任务可执行，执行 worker 销毁逻辑
        processWorkerExit(w, completedAbruptly);
    }
}

getTask()方法逻辑

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        
        // 以下两种情况递减工作线程数量
        // 1. rs >= STOP
        // 2. rs == SHUTDOWN && workQueue.isEmpty()
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        // 允许核心线程超时 或者 当前线程数 > 核心线程数，有可能发生超时关闭
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        // wc 什么情况 > maximumPoolSize ，调用 setMaximumPoolSize()方法将 maximumPoolSize 调小了，会发生这种情况，此时需要关闭多余线程
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            // 阻塞队列获取任务
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            // 发生中断，进行重试
            timedOut = false;
        }
    }
}

以上内容比较详细的介绍了 ThreadPoolExecutor 的继承体系，以及相关的核心源码，基于此，现在我们来看 DynamicTp 提供的告警通知能力。

核心参数变更通知

对应配置中心的监听端监听到配置变更后，封装到 DtpProperties 中然后交由 DtpRegistry 类中的 refresh()方法去做配置更新，同时通知时会高亮显示有变更的字段

线程池活跃度告警

活跃度 = activeCount / maximumPoolSize

服务启动后会开启一个定时监控任务，每隔一定时间（可配置）去计算线程池的活跃度，达到配置的 threshold 阈值后会触发一次告警，告警间隔内多次触发不会发送告警通知

队列容量告警

容量使用率 = queueSize / queueCapacity

服务启动后会开启一个定时监控任务，每隔一定时间去计算任务队列的使用率，达到配置的 threshold 阈值后会触发一次告警，告警间隔内多次触发不会发送告警通知

拒绝策略告警

/**
 * Do sth before reject.
 * @param executor ThreadPoolExecutor instance
 */
default void beforeReject(ThreadPoolExecutor executor) {
    if (executor instanceof DtpExecutor) {
        DtpExecutor dtpExecutor = (DtpExecutor) executor;
        dtpExecutor.incRejectCount(1);
        Runnable runnable = () -> AlarmManager.doAlarm(dtpExecutor, REJECT);
        AlarmManager.triggerAlarm(dtpExecutor.getThreadPoolName(), REJECT.getValue(), runnable);
    }
}

线程池线程数达到配置的最大线程数，且任务队列已满，再提交任务会触发拒绝策略。DtpExecutor 线程池用到的 RejectedExecutionHandler 是经过动态代理包装过的，在执行具体的拒绝策略之前会执行 RejectedAware 类 beforeReject()方法，此方法会去做拒绝数量累加（总数值累加、周期值累加）。且判断如果周期累计值达到配置的阈值，则会触发一次告警通知（同时重置周期累加值为 0 及上次告警时间为当前时间），告警间隔内多次触发不会发送告警通知

任务队列超时告警

重写 ThreadPoolExecutor 的 execute()方法和 beforeExecute()方法，如果配置了执行超时或排队超时值，则会用 DtpRunnable 包装任务，同时记录任务的提交时间 submitTime ，beforeExecute 根据当前时间和 submitTime 的差值就可以计算到该任务在队列中的等待时间，然后判断如果差值大于配置的 queueTimeout 则累加排队超时任务数量（总数值累加、周期值累加）。且判断如果周期累计值达到配置的阈值，则会触发一次告警通知（同时重置周期累加值为 0 及上次告警时间为当前时间），告警间隔内多次触发不会发送告警通知

@Override
public void execute(Runnable command) {
    if (CollUtil.isNotEmpty(taskWrappers)) {
        for (TaskWrapper t : taskWrappers) {
            command = t.wrap(command);
        }
    }

    if (runTimeout > 0 || queueTimeout > 0) {
        command = new DtpRunnable(command);
    }
    super.execute(command);
}

@Override
protected void beforeExecute(Thread t, Runnable r) {
    if (!(r instanceof DtpRunnable)) {
        super.beforeExecute(t, r);
        return;
    }
    DtpRunnable runnable = (DtpRunnable) r;
    long currTime = System.currentTimeMillis();
    if (runTimeout > 0) {
        runnable.setStartTime(currTime);
    }
    if (queueTimeout > 0) {
        long waitTime = currTime - runnable.getSubmitTime();
        if (waitTime > queueTimeout) {
            queueTimeoutCount.incrementAndGet();
            Runnable alarmTask = () -> AlarmManager.doAlarm(this, QUEUE_TIMEOUT);
            AlarmManager.triggerAlarm(this.getThreadPoolName(), QUEUE_TIMEOUT.getValue(), alarmTask);
        }
    }

    super.beforeExecute(t, r);
}

任务执行超时告警

重写 ThreadPoolExecutor 的 afterExecute()方法，根据当前时间和 beforeExecute()中设置的 startTime 的差值即可算出任务的实际执行时间，然后判断如果差值大于配置的 runTimeout 则累加排队超时任务数量（总数值累加、周期值累加）。且判断如果周期累计值达到配置的阈值，则会触发一次告警通知（同时重置周期累加值为 0 及上次告警时间为当前时间），告警间隔内多次触发不会发送告警通知

@Override
protected void afterExecute(Runnable r, Throwable t) {

    if (runTimeout > 0) {
        DtpRunnable runnable = (DtpRunnable) r;
        long runTime = System.currentTimeMillis() - runnable.getStartTime();
        if (runTime > runTimeout) {
            runTimeoutCount.incrementAndGet();
            Runnable alarmTask = () -> AlarmManager.doAlarm(this, RUN_TIMEOUT);
            AlarmManager.triggerAlarm(this.getThreadPoolName(), RUN_TIMEOUT.getValue(), alarmTask);
        }
    }

    super.afterExecute(r, t);
}

告警通知相关配置项

如果想使用通知告警功能，配置文件必须要配置 platforms 字段，且可以配置多个平台，如钉钉、企微等； notifyItems 配置具体告警项，包括阈值、平台、告警间隔等。

spring:
  dynamic:
    tp:
      # 省略其他项
      platforms:                         # 通知平台
        - platform: wechat
          urlKey: 38a98-0c5c3b649c
          receivers: test
        - platform: ding
          urlKey: f80db3e801d593604f4a08dcd6a
          secret: SECb5444a6f375d5b9d21
          receivers: 17811511815
      executors:                                   # 动态线程池配置，都有默认值，采用默认值的可以不配置该项，减少配置量
        - threadPoolName: dtpExecutor1
          executorType: common                          # 线程池类型 common 、eager：适用于 io 密集型
          corePoolSize: 2
          maximumPoolSize: 4
          queueCapacity: 200
          queueType: VariableLinkedBlockingQueue       # 任务队列，查看源码 QueueTypeEnum 枚举类
          rejectedHandlerType: CallerRunsPolicy        # 拒绝策略，查看 RejectedTypeEnum 枚举类
          keepAliveTime: 50
          allowCoreThreadTimeOut: false
          threadNamePrefix: dtp1                         # 线程名前缀
          waitForTasksToCompleteOnShutdown: false        # 参考 spring 线程池设计
          awaitTerminationSeconds: 5                     # 单位（ s ）
          preStartAllCoreThreads: false                  # 是否预热核心线程，默认 false
          runTimeout: 200                                # 任务执行超时阈值，目前只做告警用，单位（ ms ）
          queueTimeout: 100                              # 任务在队列等待超时阈值，目前只做告警用，单位（ ms ）
          taskWrapperNames: ["ttl"]                      # 任务包装器名称，集成 TaskWrapper 接口
          notifyItems:                     # 报警项，不配置自动会按默认值配置（变更通知、容量报警、活性报警、拒绝报警、任务超时报警）
            - type: capacity               # 报警项类型，查看源码 NotifyTypeEnum 枚举类
              threshold: 80                # 报警阈值
              platforms: [ding,wechat]     # 可选配置，不配置默认拿上层 platforms 配置的所以平台
              interval: 120                # 报警间隔（单位：s ）
            - type: change
            - type: liveness
              threshold: 80
              interval: 120
            - type: reject
              threshold: 1
              interval: 160
            - type: run_timeout
              threshold: 1
              interval: 120
            - type: queue_timeout
              threshold: 1
              interval: 140