线程池版流水线模式技术笔记-开发者社区

一、模式核心思想

流水线模式本质是任务分阶段串行处理，把一个完整业务任务拆分成多道独立工序（本例拆分为 TaskA、TaskB、TaskC 三个阶段）。
每个阶段由独立线程池负责消费处理，上一阶段处理完成后，自动把任务提交给下一阶段线程池，形成生产者→阶段A→阶段B→阶段C→任务收尾的流水线链路。
// 核心特点：工序解耦、各阶段线程池独立管控、任务异步串行流转、支持高并发持续生产任务

二、整体结构分层

任务实体类 Task：封装业务数据 + 各阶段业务处理逻辑
阶段任务处理类 TaskA/TaskB/TaskC：实现 Runnable，绑定对应业务阶段，处理完自动投递下一线程池
主线程启动类 Main：初始化多组独立线程池、任务队列、完成任务集合；启动生产者线程持续生成任务；启动监控线程观测队列积压与任务计算正确性

三、核心代码实现

1. 任务实体类 Task

封装任务核心变量 num，提供三段流水线处理方法，每个方法模拟业务耗时。

packageduoxiancheng.xq0529;publicclassTask{// 任务核心运算数据intnum;// 流水线第一阶段处理publicvoidtaskA(){try{// 模拟业务处理耗时500msThread.sleep(500);}catch(InterruptedExceptione){thrownewRuntimeException(e);}// 阶段A业务逻辑：数值+20num+=20;}// 流水线第二阶段处理publicvoidtaskB(){try{// 模拟业务处理耗时500msThread.sleep(500);}catch(InterruptedExceptione){thrownewRuntimeException(e);}// 阶段B业务逻辑：数值*10num*=10;}// 流水线第三阶段处理publicvoidtaskC(){try{// 模拟业务处理耗时500msThread.sleep(500);}catch(InterruptedExceptione){thrownewRuntimeException(e);}// 阶段C业务逻辑：数值平方num*=num;}}

// 备注：三个方法严格串行依赖，必须按 A→B→C 顺序执行，最终预期结果：(0+20)*10 再平方 = 40000

2. 阶段任务处理器

2.1 第一阶段 TaskA

执行完 taskA 后，将任务交给第二阶段线程池

packageduoxiancheng.xq0529;importjava.util.ArrayList;importjava.util.concurrent.Executor;// 流水线第一阶段任务publicclassTaskAimplementsRunnable{ExecutorexecutorB;// 绑定第二阶段线程池ExecutorexecutorC;// 绑定第三阶段线程池ArrayList<Task>taskDoneList;// 存放最终完成的任务Tasktask;// 当前待处理任务publicTaskA(ExecutorexecutorB,ExecutorexecutorC,ArrayList<Task>taskDoneList,Tasktask){this.executorB=executorB;this.executorC=executorC;this.taskDoneList=taskDoneList;this.task=task;}@Overridepublicvoidrun(){// 执行第一阶段业务逻辑task.taskA();// 流转：提交任务给第二阶段线程池executorB.execute(newTaskB(task,executorC,taskDoneList));}}

2.2 第二阶段 TaskB

执行完 taskB 后，将任务交给第三阶段线程池

packageduoxiancheng.xq0529;importjava.util.ArrayList;importjava.util.concurrent.Executor;// 流水线第二阶段任务classTaskBimplementsRunnable{Tasktask;ExecutorexecutorC;ArrayList<Task>taskDoneList;publicTaskB(Tasktask,ExecutorexecutorC,ArrayList<Task>taskDoneList){this.task=task;this.executorC=executorC;this.taskDoneList=taskDoneList;}@Overridepublicvoidrun(){// 执行第二阶段业务逻辑task.taskB();// 流转：提交任务给第三阶段线程池executorC.execute(newTaskC(task,taskDoneList));}}

2.3 第三阶段 TaskC

最后阶段处理完毕，把任务加入完成集合

packageduoxiancheng.xq0529;importjava.util.ArrayList;// 流水线第三阶段任务classTaskCimplementsRunnable{Tasktask;ArrayList<Task>taskDoneList;publicTaskC(Tasktask,ArrayList<Task>taskDoneList){this.task=task;this.taskDoneList=taskDoneList;}@Overridepublicvoidrun(){// 执行第三阶段业务逻辑task.taskC();// 整个流水线处理完成，加入完成列表taskDoneList.add(task);}}

// 备注：各阶段通过构造器传递下一级线程池与任务上下文，实现任务自动流转，无需手动阻塞等待

3. 主线程工厂启动类 Main

初始化三级线程池、队列，启动生产者持续造任务，启动监控线程观测运行状态。

packageduoxiancheng.xq0529;importjava.util.ArrayList;importjava.util.concurrent.ArrayBlockingQueue;importjava.util.concurrent.Executor;importjava.util.concurrent.ThreadPoolExecutor;importjava.util.concurrent.TimeUnit;publicclassMain{publicstaticvoidmain(String[]args){// 为每个阶段独立定义阻塞队列，容量20ArrayBlockingQueue<Runnable>queueA=newArrayBlockingQueue<>(20);ArrayBlockingQueue<Runnable>queueB=newArrayBlockingQueue<>(20);ArrayBlockingQueue<Runnable>queueC=newArrayBlockingQueue<>(20);// 保存所有流水线执行完成的任务ArrayList<Task>taskDoneList=newArrayList<>();// 初始化三级独立线程池 // 核心线程5、最大线程10、空闲超时1000msExecutorexecutorA=newThreadPoolExecutor(5,10,1000,TimeUnit.MILLISECONDS,queueA);ExecutorexecutorB=newThreadPoolExecutor(5,10,1000,TimeUnit.MILLISECONDS,queueB);ExecutorexecutorC=newThreadPoolExecutor(5,10,1000,TimeUnit.MILLISECONDS,queueC);// 生产者线程：循环不断生成新任务，提交给第一阶段线程池newThread(()->{intcount=0;while(true){try{// 控制生产速率Thread.sleep(50);}catch(InterruptedExceptione){thrownewRuntimeException(e);}// 新建任务实例Tasktask=newTask();// 提交到流水线首阶段executorA.execute(newTaskA(executorB,executorC,taskDoneList,task));count++;System.out.println("生产者已生成任务数:"+count);}}).start();// 监控线程：定时查看队列积压、完成任务数、计算正确率newThread(()->{while(true){try{// 每秒监控一次Thread.sleep(1000);}catch(InterruptedExceptione){thrownewRuntimeException(e);}// 打印三个阶段队列积压大小System.out.println("队列A大小:"+queueA.size()+" 队列B大小:"+queueB.size()+" 队列C大小:"+queueC.size());// 统计结果正确的任务数量intcorrectCount=0;for(inti=0;i<taskDoneList.size();i++){// 符合预期结果40000即为正确if(taskDoneList.get(i).num==40000){correctCount++;}}System.out.println("已完成任务总数:"+taskDoneList.size()+" 运算正确任务数:"+correctCount);}}).start();}}