文章内容
AsyncTaskPool 是一个轻量多线程框架，其核心代码不到 400 行，同时支持三种常用设计模式。
一、整体架构：三层角色
AsyncTaskPool 只有三个核心角色：
- BasicTaskDispatchPool——对外的门面。通过继承实现三个抽象方法：
runTask（具体任务逻辑）、getFinishedCallback（全部完成后回调）、getItemTaskCallback（单个任务完成后回调），负责接收任务并转交给调度层。 - ThreadDisPatchManager——调度中枢。管理线程的创建与销毁、任务的三级分发、中断信号的处理。所有并发控制的细节都在这里。
- JobThread / JobTask——执行单元。每个
JobThread是一条线程，内部跑一个JobTask（Runnable实现）。JobTask持有自己的任务队列，逐个取出执行。
三个角色的职责边界：门面负责暴露 API，调度层负责决策，执行层负责干活。
二、模板方法：定义骨架，抽象方法
BasicTaskDispatchPool 应用了模板方法模式——把线程池如何运作的流程固化，把具体任务做什么和完成后如何通知留给子类决定。
java
// 泛型参数 T=输入类型，O=输出类型// 实现 IThreadContainer 暴露线程池的运行时状态（线程数、队列长度等）public abstract class BasicTaskDispatchPool<T, O> implements IThreadContainer { // 默认每个线程最多缓存 10 个待处理任务 private static final int JOB_MAX_TASKS = 10; // 默认线程数 = CPU 核数，让每个核心跑一个线程 private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors(); // 内部持有的调度管理器，真正的并发控制都在这个对象里 private ThreadDisPatchManager<T, O> mThreadPool; // 构造函数：把子类实现的三个抽象方法以回调形式注入调度管理器 // 注意 runTask 用 lambda 桥接了一层，确保 this 指向的是子类实例 protected BasicTaskDispatchPool() { mThreadPool = new ThreadDisPatchManager<T, O>( getMaxThread(), // 最大线程数（可被子类覆写） getMaxSingleThreadTask(), // 单线程最大任务数（可被子类覆写） arg -> BasicTaskDispatchPool.this.runTask(arg), // 将 runTask 包装为 JobDetail getFinishedCallback(), // 全部完成的回调 getItemTaskCallback() // 单个任务完成的回调 ); } // 对外暴露的唯一入口：提交一个任务 // createWhenIdle 这个名字暗示了"有空闲就创建/分配，否则阻塞" public BasicTaskDispatchPool addTask(T task) { if (task != null) { mThreadPool.createWhenIdle(task); } return this; // 返回 this 支持链式调用 } // 以下两个是 protected，允许子类覆写以自定义线程/队列容量 protected int getMaxThread() { return CPU_COUNT; } protected int getMaxSingleThreadTask() { return JOB_MAX_TASKS; } // ----- 三个抽象方法，子类必须实现 ----- // ① 核心任务逻辑：输入 T，输出 O，由工作线程调用 protected abstract O runTask(T arg); // ② 全部任务完成回调：收到两个列表，分别是被完成的和被中断的 protected abstract ThreadDisPatchManager.ThreadTaskFinished<T, O> getFinishedCallback(); // ③ 单个任务完成回调：分别处理 onInterrupted 和 onFinished 两种情况 protected abstract ThreadDisPatchManager.JobTaskFinished<T, O> getItemTaskCallback();}构造函数里的默认策略：线程数取 CPU 核数，每个线程的任务队列上限为 10。getMaxThread() 和 getMaxSingleThreadTask() 都是 protected，子类可以选择覆写。比如背压版本的 Demo 把单线程队列上限从 10 改为 2，降低消耗效率。
使用者只需要实现三个方法：
java
// 泛型绑定：输入 InputArgs，输出 Responsepublic class DemoAsyncPool extends BasicTaskDispatchPool<InputArgs, Response> { // ① 工作线程调用：收到 arg，执行逻辑，返回结果 @Override protected Response runTask(InputArgs arg) { // 实际业务逻辑：这里模拟耗时操作 return new Response(Thread.currentThread().toString()); } // ② 所有线程都空闲且阻塞队列也空了时调用 @Override protected ThreadTaskFinished<InputArgs, Response> getFinishedCallback() { return (finishedList, interruptedList) -> { // finishedList：正常完成的任务及结果 // interruptedList：执行了但因中断标记被分类到此的任务 System.out.println("全部完成: " + finishedList.size()); }; } // ③ 每个任务执行完毕时调用（无论是否被中断） @Override protected JobTaskFinished<InputArgs, Response> getItemTaskCallback() { return new JobTaskFinished<>() { // 任务执行完但线程被标记为中断 public void onInterrupted(InputArgs args) { /* ... */ } // 任务正常完成，收到输入-输出的配对结果 public void onFinished(Pair<InputArgs, Response> res) { /* ... */ } }; }}模板方法的核心：
- 不变的流程封装在基类，可变的行为推迟到子类。
- 线程的创建、分配、回收是不变。
- 任务的具体逻辑和结果通知可变。
三、调度核心：三级分发算法
所有调度复杂度集中在 ThreadDisPatchManager.create() 方法里，用一个 synchronized 方法实现了三级分发。
3.1 第一级：线程数未满 → 创建新线程
java
// synchronized 保证整个三级分发是原子的// 当前活跃线程数还没到上限（默认 CPU 核数）if (mThreadList.size() < maxThread) { // 创建 JobTask，它实现了 Runnable，传给 Thread 执行 JobTask tJobTask = new JobTask(/* 内嵌回调，见 3.4 节 */); // 把当前任务作为 JobTask 的第一个任务 tJobTask.addJob(job); // 用 JobTask 构造 JobThread（Thread 子类），并启动 JobThread jobThread = new JobThread(tJobTask); mThreadList.add(jobThread); // 登记到线程列表 jobThread.start(); // 启动线程，进入 JobTask.run() 的死循环 return true; // 分配成功}3.2 第二级：有线程空闲 → 塞进队列
java
// 线程数已达上限，遍历找有空位的线程for (JobThread thread : mThreadList) { if (thread.isIdle()) { // isIdle(): 队列长度 < maxThreadJob thread.addNewJob(job); // 把任务追加到该线程的 taskQueue 末尾 return true; // 分配成功 }}3.3 第三级：全部满载 → 进入全局阻塞队列
java
// 所有线程队列都满了，任务进入全局阻塞队列等待if (blockingJobs.contains(job)) { // 防重复：同一个任务对象不重复入队 System.out.println("repeat task!! :" + job);} else { blockingJobs.add(job); // ConcurrentLinkedQueue，线程安全}return false; // 返回 false 表示任务未立即分配，进入了等待状态3.4 自调度：线程干完活后的链式反应
任务提交后，当某个线程把自己的任务队列处理干净时，它触发一次自调度。这个机制的入口在 JobTaskFinished 回调：
java
// 这段代码位于 ThreadDisPatchManager.create() 方法内部，// 它构造了每个 JobTask 的回调对象public void onFinished(Pair<T, O> res) { // 判断是不是"队列已空"的特殊标记 if (res instanceof InnerSingleQueueFinishIndicator) { Thread thread = Thread.currentThread(); thread.interrupt(); // 标记当前线程为中断状态 mThreadList.remove(thread); // 从线程列表中注销 } else { // 正常的单任务完成，转发给用户注册的回调 mJobTaskFinished.onFinished(res); } // 关键：无论哪种情况，都触发一次 create() // 线程退出后自动检查阻塞队列，实现了"自驱式调度" create();}无参 create() 的逻辑：
java
private void create() { // 阻塞队列里还有排队任务 → 取一个出来走三级分发（回到 3.1） if (!blockingJobs.isEmpty()) { create(blockingJobs.poll()); } else { // 阻塞队列也空了，且没有存活的线程 // → 意味着所有任务都处理完毕，触发"全部完成"回调 if (livingThreadSize() == 0) { callback.onFinished(mFinishedPairs, mInterruptPairs); } }}两个关键动作：阻塞队列还有任务 → 取出来走三级分发；阻塞队列也空了、且没有活着的线程 → 触发 callback.onFinished()，通知全部任务完成。这个自驱式链路通常只在有事可做的时刻触发。
四、线程内部Loop
JobTask 实现了 Runnable，它的 run() 方法是整个执行模型的核心：
java
@Overridepublic void run() { while (true) { // 死循环，直到队列为空才退出 T arg = null; if (!taskQueue.isEmpty()) { // ① 队列非空 → 取出一个任务 arg = taskQueue.poll(); // 从队头取出任务（FIFO 保证公平） // ② 执行用户定义的业务逻辑 Pair<T, O> toPair = new Pair<>(arg, runTask(arg)); // ③ 检查中断标记，分类存储结果 if (!Thread.interrupted()) { // 未被中断 → 记入完成列表 mFinishedPairs.add(toPair); innerJobTaskFinished.onFinished(toPair); } else { // 已被中断 → 记入中断列表 mInterruptPairs.add(toPair); innerJobTaskFinished.onInterrupted(arg); } } else { // ④ 队列已空 → 准备退出 // 防御性检查：理论上此时队列应为空，如果不空且线程被中断则抛异常 if (!taskQueue.isEmpty() && Thread.interrupted()) { throw new RuntimeException("taskQueue is not empty:" + taskQueue.size()); } removeAll(); // 清空队列残留 // 发出特殊标记 InnerSingleQueueFinishIndicator // 它继承自 Pair，被回调识别后触发线程注销和自调度（见 3.4） innerJobTaskFinished.onFinished( new InnerSingleQueueFinishIndicator(null, null)); break; // 退出死循环，run() 结束，线程自然终止 } }}具体解释：
- 从自己的队列取任务（
taskQueue.poll()）——FIFO 顺序，保证先提交的先处理。 - 执行
runTask(arg)——调用用户覆写的业务逻辑，同步返回结果。 - 检查
Thread.interrupted()——任务执行完后再检查中断标记。被中断的任务结果同样生成，只是分类到mInterruptPairs，避免数据丢失。 - 队列空了 → 发出
InnerSingleQueueFinishIndicator信号退出循环。这个信号在ThreadDisPatchManager的回调中被识别，触发"线程注销 + 检查阻塞队列"的自调度链。
线程不是被外部杀死的——它自己执行完队列的所有任务后主动退出循环。整个流程：队列清空 → 线程主动结束 → 从管理器注销 → 检查阻塞队列有无剩余任务。
五、背压：策略模式的实现
当所有线程的队列都满、阻塞队列也在堆积时怎么办？AsyncTaskPool 把满了如何处理的决策抽象成一个策略接口：
java
public interface IBackPressureStrategy { // 全部线程满负荷运转时调用 // maxThreadLimit: 最大线程上限 livingThread: 当前存活线程数 blockingJobs: 阻塞队列长度 void onFullBusy(int maxThreadLimit, int livingThread, int blockingJobs); // 有空闲线程时调用 // busyRate: 繁忙率 = livingThread / maxThreadLimit void onIDle(float busyRate, int maxThreadLimit, int livingThread);}BackPressureTaskDispatchPool 同时继承 BasicTaskDispatchPool 和实现 IBackPressureStrategy：
java
// 既是线程池（extends BasicTaskDispatchPool），也是背压策略（implements IBackPressureStrategy）public abstract class BackPressureTaskDispatchPool<T, O> extends BasicTaskDispatchPool<T, O> implements IBackPressureStrategy { // ① 生产者侧：每次 addTask 时触发策略检查 @Override public BasicTaskDispatchPool addTask(T task) { strategyInvoke(); // 提交任务前先感知当前压力 return super.addTask(task); // 然后走正常的调度流程 } // ② 消费者侧：每次 runTask 时触发策略检查 // 这里返回 null 而非真正结果——实际的任务逻辑在策略调用之后由使用者决定 @Override protected O runTask(T arg) { strategyInvoke(); // 执行任务前也感知一次压力 return null; // 基类的方法体，子类需要 override 来执行真正的任务 } // 统一的状态检测与策略分发 private void strategyInvoke() { int blockingJobs = getBlockingJobs(); // 全局阻塞队列中的任务数 int livingThread = getLivingThread(); // 当前存活的线程数 int maxThreadLimit = getMaxThreadLimit(); // 最大线程上限 float busyRate = (float) livingThread / (float) maxThreadLimit; if (maxThreadLimit == livingThread) { // 线程全满 → 通知策略"满载" onFullBusy(maxThreadLimit, livingThread, blockingJobs); } else { // 还有空闲 → 通知策略"宽松" onIDle(busyRate, maxThreadLimit, livingThread); } }}策略在两处触发——addTask 在生产者提交任务前检查状态，runTask 在消费者执行任务前检查状态。两端感知压力，保证无论哪一侧先达到瓶颈，策略都能被调起。
Demo 中的策略实现：
java
@Overridepublic void onFullBusy(int maxThreadLimit, int livingThread, int blockingJobs) { isBusying = true; // 标记"忙"，外部提交者读这个标志决定是否降速 System.out.println("Stop! Is full busing");} @Overridepublic void onIDle(float busyRate, int maxThreadLimit, int livingThread) { isBusying = false; // 标记"闲"，外部提交者恢复正常提交 System.out.println("Idle, Come on, faster!");}生产者在提交任务前检查状态：
java
// 外部提交循环if (demoAsyncPool.isBusying()) { Thread.sleep(10000); // 线程池满，生产者休眠 10 秒再试}demoAsyncPool.addTask(new InputArgs("job:" + i)); // 空闲时继续提交背压的本质不是拒绝任务，而是通知生产者慢一点。比直接抛出异常更友好——生产者可以自行决定降速策略。子类覆写 onFullBusy 和 onIDle 可以实现任意行为：打日志、发告警、限流、甚至动态扩容。核心调度逻辑不动，策略随意替换。
六、线程安全
数据结构选型。 mThreadList 用 CopyOnWriteArrayList——线程列表主要在遍历找空闲线程，写入只在创建和注销时发生，读远多于写。blockingJobs 则是用线程安全的 ConcurrentLinkedQueue。
create() 方法是 synchronized 的。 整个三级分发在一个同步块里完成，避免刚检查完空闲、任务就被另一个线程塞进来的竞态。这个同步块只做检查和分配，不包含执行任务的部分，一般不会成为性能瓶颈。
中断语义。 被中断的任务不丢失结果——runTask(arg) 执行完后，根据中断标记将结果分入 mFinishedPairs 或 mInterruptPairs。回调对两批结果分别处理，不做模糊合并。
线程退出是自驱动的。 线程执行完队列中所有任务后 break 退出循环，回调中完成注销，整个清理路径在同步控制流内完成。
七、设计模式总结
三种设计模式：
- 模板方法（
BasicTaskDispatchPool）定义了线程池如何运作的骨架——线程创建、任务分配、结果回调。 - 策略模式（
IBackPressureStrategy）让满了之后（背压）怎么办成为一个可替换的决策点，不影响调度骨架。 - 回调链（
JobTaskFinished、ThreadTaskFinished）构成两层通知机制：单任务完成通知和全部完成通知，同时承担了自调度的触发职责——线程退出后通过回调自动检查阻塞队列。
本文由 AgentPlanFlow 生成