java 多线程之Worker Thread模式(Thread Pool模式)

饭宝  论坛元老 | 7 天前 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 1641|帖子 1641|积分 4923

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

x
Worker Thread模式

Worker的意思是工作的人，在Worker Thread模式中，工人线程Worker thread会逐个取回工作并进行处置惩罚，当所有工作全部完成后，工人线程会期待新的工作到来。
Worker Thread模式也被成为Background Thread（配景线程）模式，另外，如果从生存多个工人线程的场合这一点看，我们也可以称这种模式为Thread Pool模式。
Worker Thread模式中的脚色

TransPortThread [将部件塞到传送带上]（委托者）
创建表现工作请求的Cargo并将其通报给SimpleChannel 。在示例步调中，TransPortThread 相称于该脚色。
2.SimpleChannel [传送带的脚色]（通讯线路）
SimpleChannel 脚色接受来自于TransPortThread 的Cargo，并将其通报给WorkerThread。在示例步调中，SimpleChannel 相称于该脚色。
3.WorkerThread（工人）
WorkerThread脚色从Channel中获取Cargo，并进行工作，当一项工作完成后，它会继承去获取另外的Cargo，在示例步调中，WorkerThread相称于该脚色。
4.Cargo（部件）
Cargo脚色是表现工作中的部件，Cargo脚色中生存了进行工作所必须的信息，在示例步调中，Cargo相称于该部件脚色。
Worker Thread利用场景

想象这样一个场景，一个工厂在生产货物，在一个车间里，有几个工人，每次生产部件准备后，车间外的人就将部件放到车间的传送带上，工人每次做完一个货品就从桌子上取部件。在这里，注意到，部件并不是直接交给工人的，另外一点，工人并不是做完一个部件就回家换个新人，后者在现实有点滑稽，但是在步调中却对应一个典型的线程利用方法：线程池。
所谓线程池，就是对线程的复用，当线程执行完任务之后就继承取其他任务执行，而不是销毁启动新线程执行其他任务。因为线程的启动对于系统性能开销比较大，所以这样对于系统性能的进步很有好处。
Main.java

  1. public class WorkClient {
  3.     public static void main(String[] args) {
  4.         // 创建容量为5的SimpleChannel实例（数据缓冲通道）
  5.         SimpleChannel simpleChannel = new SimpleChannel(5);
  7.         // 启动工作通道开始处理数据（消费者启动）
  8.         simpleChannel.startWork();
  10.         // 启动三个数据传输线程（生产者角色）
  11.         new TransPortThread("june", simpleChannel).start();
  12.         new TransPortThread("july", simpleChannel).start();
  13.         new TransPortThread("alen", simpleChannel).start();
  14.     }
  16. }
复制代码
Cargo.java

脚色：部件或货物
  1. import java.util.Optional;
  3. /**
  4. * 货物类，包含货物编号和名称，并提供执行操作时打印线程执行信息的功能
  5. *
  6. * @author [请填写作者名称]
  7. * @version 1.0
  8. */
  9. public class Cargo {
  11.     /**
  12.      * 货物名称
  13.      */
  14.     private final String name;
  16.     /**
  17.      * 货物编号
  18.      */
  19.     private final int num;
  21.     /**
  22.      * 构造方法，初始化货物信息
  23.      *
  24.      * @param name 货物名称
  25.      * @param num  货物编号
  26.      */
  27.     public Cargo(String name, int num) {
  28.         this.name = name;
  29.         this.num = num;
  30.     }
  32.     /**
  33.      * 执行货物处理操作，打印当前线程名称和货物信息
  34.      */
  35.     public void execute(){
  36.         Optional.of(Thread.currentThread().getName()+" executed "+this.toString()).ifPresent(System.out::println);
  37.     }
  39.     /**
  40.      * 获取货物信息的字符串表示
  41.      *
  42.      * @return 格式化后的货物信息（例如：Cargo ==> NO.123 Name.ItemA）
  43.      */
  44.     @Override
  45.     public String toString() {
  46.         return "Cargo ==> NO."+num+" Name."+name;
  47.     }
  48. }
复制代码
SimpleChannel.java

职责：将部件或货物从源头传给处置惩罚端
  1. import java.util.Arrays;
  3. /**
  4. * 简单线程通信通道，管理货物队列和工作线程池
  5. * 使用环形缓冲区实现生产者-消费者模式，通过synchronized和wait/notify机制保证线程安全
  6. */
  7. public class SimpleChannel {
  9.     /** 缓冲区最大容量 */
  10.     private static final int BUFFER_SIZE = 100;
  12.     /** 环形缓冲区数组，存储货物对象 */
  13.     private final Cargo[] cargoQueue;
  15.     /** 队列头指针，指向下一个要取出的元素位置 */
  16.     private int headIndex;
  18.     /** 当前队列中的元素数量 */
  19.     private int count;
  21.     /** 队列尾指针，指向下一个要插入的元素位置 */
  22.     private int tailIndex;
  24.     /** 工作线程池 */
  25.     private final WorkerThread[] workerPool;
  27.     /**
  28.      * 构造方法初始化通道
  29.      * @param workerCount 工作线程数量
  30.      */
  31.     public SimpleChannel(int workerCount) {
  32.         this.cargoQueue = new Cargo[BUFFER_SIZE];
  33.         this.headIndex = 0;
  34.         this.count = 0;
  35.         this.tailIndex = 0;
  36.         this.workerPool = new WorkerThread[workerCount];
  37.         this.init();
  38.     }
  40.     /**
  41.      * 初始化工作线程池
  42.      * 为每个线程设置名称并绑定当前通道实例
  43.      */
  44.     private void init() {
  45.         for (int i = 0; i < workerPool.length; i++) {
  46.             workerPool[i] = new WorkerThread("worker-"+i, this);
  47.         }
  48.     }
  50.     /**
  51.      * 启动所有工作线程开始工作
  52.      * 通过遍历线程池调用start()方法启动每个线程
  53.      */
  54.     public void startWork() {
  55.         Arrays.asList(workerPool).forEach(WorkerThread::start);
  56.     }
  58.     /**
  59.      * 生产者添加货物到缓冲区
  60.      * @param cargo 要添加的货物对象
  61.      * @throws InterruptedException 线程中断异常
  62.      */
  63.     public synchronized void pushCargo(Cargo cargo) {
  64.         while (count >= BUFFER_SIZE) {
  65.             try {
  66.                 this.wait(); // 缓冲区满时等待
  67.             } catch (InterruptedException e) {
  68.                 // 异常处理（建议补充日志）
  69.             }
  70.         }
  71.         // 添加货物到尾部并更新指针
  72.         this.cargoQueue[tailIndex] = cargo;
  73.         this.tailIndex = (this.tailIndex + 1) % this.cargoQueue.length;
  74.         this.count++;
  75.         this.notifyAll(); // 通知等待线程
  76.     }
  78.     /**
  79.      * 消费者从缓冲区取出货物
  80.      * @return 取出的货物对象
  81.      * @throws InterruptedException 线程中断异常
  82.      */
  83.     public synchronized Cargo popCargo() {
  84.         while (count <= 0) {
  85.             try {
  86.                 this.wait(); // 缓冲区空时等待
  87.             } catch (InterruptedException e) {
  88.                 // 异常处理（建议补充日志）
  89.             }
  90.         }
  91.         // 取出头部元素并更新指针
  92.         Cargo cargo = this.cargoQueue[headIndex];
  93.         this.headIndex = (this.headIndex + 1) % this.cargoQueue.length;
  94.         this.count--;
  95.         this.notifyAll(); // 通知等待线程
  96.         return cargo;
  97.     }
  98. }
复制代码
TransPortThread .java

职责：将请求的信息塞到传输通道上去
  1. /**
  2. * 运输线程，负责持续生成货物并推送到传输通道 {@link SimpleChannel}
  3. */
  4. public class TransPortThread extends Thread {
  6.     private SimpleChannel simpleChannel;
  8.     /**
  9.      * 静态随机数生成器，用于生成随机休眠时间
  10.      */
  11.     private static final Random RANDOM = new Random(System.currentTimeMillis());
  13.     /**
  14.      * 构造方法
  15.      *
  16.      * @param name 线程名称
  17.      * @param simpleChannel 货物传输通道
  18.      */
  19.     public TransPortThread(String name, SimpleChannel simpleChannel) {
  20.         super(name);
  21.         this.simpleChannel = simpleChannel;
  22.     }
  24.     @Override
  25.     public void run() {
  26.         try {
  27.             int i = 0;
  28.             // 无限循环持续生成货物
  29.             while (true){
  30.                 Cargo cargo = new Cargo(getName(), i++); // 创建货物对象（包含线程名和序号）
  31.                 this.simpleChannel.pushCargo(cargo);   // 将货物推送到通道
  32.                 Thread.sleep(RANDOM.nextInt(1000));    // 随机休眠 0-1000ms
  33.             }
  34.         } catch (InterruptedException e) {
  35.             // 线程中断时的异常处理（当前空实现）
  36.         }
  37.     }
  38. }
复制代码
WorkerThread.java

职责：从传输带上获取部件或货物
  1. import java.util.Random;
  2. /**
  3. * 工作线程类，负责从共享通道中持续获取任务并执行
  4. */
  5. public class WorkerThread extends Thread {
  7.     /**
  8.      * 存储任务通道实例，用于获取待处理的任务对象
  9.      */
  10.     private final SimpleChannel simpleChannel;
  12.     /**
  13.      * 静态随机数生成器，用于生成随机休眠时间（模拟任务执行间隔）
  14.      */
  15.     private static final Random random = new Random(System.currentTimeMillis());
  17.     /**
  18.      * 构造方法初始化线程名称和任务通道
  19.      * @param name 线程名称标识
  20.      * @param simpleChannel 任务通道对象，用于获取待处理任务
  21.      */
  22.     public WorkerThread(String name, SimpleChannel simpleChannel) {
  23.         super(name);
  24.         this.simpleChannel = simpleChannel;
  25.     }
  27.     /**
  28.      * 线程执行主体方法，持续循环执行以下操作：
  29.      * 1. 从通道中获取任务对象
  30.      * 2. 执行任务
  31.      * 3. 随机休眠（模拟任务处理间隔）
  32.      */
  33.     @Override
  34.     public void run() {
  35.         while (true) {
  36.             try {
  37.                 // 从通道中获取任务对象
  38.                 Cargo cargo = simpleChannel.popCargo();
  39.                 // 执行任务逻辑
  40.                 cargo.execute();
  41.                 // 随机休眠0-1000毫秒（模拟任务间隔）
  42.                 Thread.sleep(random.nextInt(1000));
  43.             } catch (Exception e) {
  44.                 // 异常处理：打印堆栈信息（实际应用中建议添加更完善的异常处理）
  45.                 e.printStackTrace();
  46.             }
  47.         }
  48.     }
  49. }
复制代码
执行的效果
  1. worker-1 executed Cargo ==> NO.0 Name.alen
  2. worker-0 executed Cargo ==> NO.0 Name.june
  3. worker-4 executed Cargo ==> NO.0 Name.july
  4. worker-3 executed Cargo ==> NO.1 Name.june
  5. worker-0 executed Cargo ==> NO.1 Name.alen
  6. worker-2 executed Cargo ==> NO.1 Name.july
  7. worker-1 executed Cargo ==> NO.2 Name.june
  8. worker-0 executed Cargo ==> NO.2 Name.july
  9. worker-2 executed Cargo ==> NO.2 Name.alen
  10. worker-3 executed Cargo ==> NO.3 Name.june
  11. worker-4 executed Cargo ==> NO.4 Name.june
  12. worker-2 executed Cargo ==> NO.3 Name.alen
  13. worker-3 executed Cargo ==> NO.3 Name.july
  14. worker-4 executed Cargo ==> NO.5 Name.june
  15. worker-1 executed Cargo ==> NO.6 Name.june
  16. worker-0 executed Cargo ==> NO.4 Name.alen
  17. worker-2 executed Cargo ==> NO.7 Name.june
  18. worker-3 executed Cargo ==> NO.5 Name.alen
  19. worker-2 executed Cargo ==> NO.4 Name.july
  20. worker-4 executed Cargo ==> NO.6 Name.alen
  21. worker-0 executed Cargo ==> NO.5 Name.july
  22. worker-2 executed Cargo ==> NO.8 Name.june
  23. worker-3 executed Cargo ==> NO.7 Name.alen
  24. worker-1 executed Cargo ==> NO.6 Name.july
  25. worker-4 executed Cargo ==> NO.8 Name.alen
  26. worker-4 executed Cargo ==> NO.9 Name.june
  27. worker-2 executed Cargo ==> NO.10 Name.june
  28. worker-3 executed Cargo ==> NO.9 Name.alen
  29. worker-1 executed Cargo ==> NO.7 Name.july
  30. worker-2 executed Cargo ==> NO.10 Name.alen
  31. worker-4 executed Cargo ==> NO.11 Name.alen
  32. worker-3 executed Cargo ==> NO.11 Name.june
  33. worker-0 executed Cargo ==> NO.8 Name.july
  34. worker-4 executed Cargo ==> NO.12 Name.alen
  35. worker-3 executed Cargo ==> NO.12 Name.june
  36. worker-4 executed Cargo ==> NO.13 Name.alen
  37. worker-1 executed Cargo ==> NO.9 Name.july
  38. worker-2 executed Cargo ==> NO.14 Name.alen
  39. worker-0 executed Cargo ==> NO.13 Name.june
  40. worker-4 executed Cargo ==> NO.15 Name.alen
  41. worker-2 executed Cargo ==> NO.10 Name.july
  42. worker-1 executed Cargo ==> NO.14 Name.june
  43. worker-3 executed Cargo ==> NO.15 Name.june
  44. worker-4 executed Cargo ==> NO.16 Name.alen
  45. worker-3 executed Cargo ==> NO.11 Name.july
  46. worker-2 executed Cargo ==> NO.12 Name.july
  47. worker-0 executed Cargo ==> NO.16 Name.june
  48. worker-1 executed Cargo ==> NO.17 Name.june
  49. worker-2 executed Cargo ==> NO.18 Name.june
  50. worker-2 executed Cargo ==> NO.17 Name.alen
  51. worker-4 executed Cargo ==> NO.13 Name.july
  52. worker-0 executed Cargo ==> NO.18 Name.alen
  53. worker-0 executed Cargo ==> NO.19 Name.june
  54. worker-3 executed Cargo ==> NO.14 Name.july
  55. worker-1 executed Cargo ==> NO.19 Name.alen
  56. worker-2 executed Cargo ==> NO.20 Name.june
  57. worker-3 executed Cargo ==> NO.21 Name.june
  58. worker-0 executed Cargo ==> NO.15 Name.july
  59. worker-3 executed Cargo ==> NO.20 Name.alen
  60. worker-4 executed Cargo ==> NO.21 Name.alen
  61. worker-3 executed Cargo ==> NO.22 Name.june
  62. worker-1 executed Cargo ==> NO.16 Name.july
  63. worker-2 executed Cargo ==> NO.17 Name.july
  64. worker-0 executed Cargo ==> NO.23 Name.june
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

饭宝

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表