devbox
羊村懒王
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

[Java Web]定时器（Timer） 线程池_timer线程池

作者：羊村懒王 | 2024-04-27 07:37:11

赞

踩

timer线程池

使用 Timer

  1. // 使用定时器
  2. public class UseTimer {
  3.     public static void main(String[] args) {
  4.         // 一个定时器 + n 个定时器要去执行的任务
  5.         Timer timer = new Timer();   // 闹钟
  6.  
  7.         // 继承 TimerTask（抽象类）
  8.         // 执行的是 task.run() 方法
  9.         TimerTask task = new TimerTask() {
  10.             @Override
  11.             public void run() {
  12.                 System.out.println("执行任务");
  13.             }
  14.         };
  15.         TimerTask t2 = new TimerTask() {
  16.             @Override
  17.             public void run() {
  18.                 System.out.println("另一个任务");
  19.             }
  20.         };
  21.  
  22.         // 指定什么去执行
  23.         // delay: 延时，单位是 ms
  24.         // 一次性的
  25.         //timer.schedule(task, 5000);
  26.  
  27.         // fixed: 固定的
  28.         // rate: 频率
  29.         // period: 周期
  30.         timer.scheduleAtFixedRate(task, 5000, 1000);
  31.  
  32.         timer.schedule(t2, 3000);
  33.         
  34.         while(true); // 这个死循环的目的是延时，随后的任务不是在主线程中执行
  35.     }
  36. }

实现 Timer

如何实现Timer：

1.每个任务创建一个线程 + 休眠方式         缺点：每一个任务都需要关联一个线程

  1. import java.util.concurrent.TimeUnit;
  2.  
  3. public class ImplTimer {
  4.     static class MyThread extends Thread{
  5.         @Override
  6.         public void run() {
  7.             try {
  8.                 TimeUnit.SECONDS.sleep(5);
  9.             } catch (InterruptedException e) {
  10.                 e.printStackTrace();
  11.             }
  12.             while (true){
  13.                 System.out.println("执行任务");
  14.                 try {
  15.                     TimeUnit.SECONDS.sleep(1);
  16.                 } catch (InterruptedException e) {
  17.                     e.printStackTrace();
  18.                 }
  19.             }
  20.  
  21.         }
  22.     }
  23.  
  24.     public static void main(String[] args) {
  25.         MyThread thread = new MyThread();
  26.         thread.start();
  27.  
  28.         while (true);
  29.     }
  30. }

2.官方做法：一个线程，完成所有任务的执行

生产者（schedule 的线程） - 消费者（Worker）

放入队列 t1(delay = 5)、t2(delay = 3)、t3(delay = 1)

取任务，可能要等待一段时间、然后执行任务，t1、t2、t3 应该先取出 delay 最小的，这是一个优先级阻塞队列

但这个 delay 不断的在变化，因为前面 t3 等了 1s 之后，那么  t2 再等 2s 就可以执行了，这样等待的时间一直是变化的，因此我们 delay = runAt（任务运行时间（不会变）） - now（当前时间（一直在变））。我们取出 runAt 最小的。

流程：

1.准备好义格优先级队列（要在P/C两个线程中共享）

2.准备好一个等待/通知 对象（要在P/C两个线程中共享）

3.创建好一个 Worker 线程（传入 队列 + 通知对象），并启动

P线程：

  1. 根据 delay + now 计算 runAt
  2. 将 任务（TimerTask）放入队列
  3. 利用通知对象通知（通知的是 C 中的 3）

C线程：

  1. 获取最近的任务（TimerTask）
  2. 得到最近任务的延迟 delay' = runAt - now'
  3. 带条件的休眠
  4. 醒来之后，如果任务时间已到，执行任务；否则，先把任务放回去
  5. 循环回 1 继续

sleep VS wait（timeout)

sleepwait（timeout）
休眠带超时的等待
结束条件唯一

结束条件两个：

1.timeout已到      2.被唤醒

和锁没关系（不会释放）

释放锁（只会释放当前对象的锁）

Thread staticObject

线程池（Thread Pool）

主线程读取输入                                                有客人点菜

创建线程计算                                                    雇一个厨师 做饭

        创建线程 + 计算 + 销毁线程                      解雇厨师

生产者——消费者模型（多消费者）

1.Executor接口        一种执行任务的东西/人

        void execute（Runnable command);

2.ExecutorService接口

3.ThreadPoolExecutor实现类

        重点：构造方法的参数理解、创建线程的流程

线程池：公司       

正式工：一种工作线程、永远不会被“解雇”

临时工：一种工作线程、临时工空闲太久会被解雇

int corePoolSize        正式员工的数量上限（编制）

int maximumPoolSize        所有员工的数量上限（编制） 正式 + 临时

long keepAliveTime + TimeUnit unit :允许临时工摸鱼多久

BlockingQueue workQueue：传递任务的阻塞队列

ThreadFactory factory：让我们可以介入线程的创建流程

Reject(拒绝) ExecutionHandler：拒绝策略

 4.线程池的流程如下（按需创建）

  1. 一开始公司一个员工都没有
  2. 当有任务来了之后，当前正式工的数量 < 正式工的上限 一律招一个新的正式员工处理业务（即使有其他人还在空闲中）、把任务放入阻塞队列中、如果阻塞队列满了，雇佣临时工执行任务（临时工的数量 < 临时工的上限）
  3. 处理不过来执行拒绝策略

5.怎么让摸鱼的正式员工也被解雇？

shutdown()    VS    shutdownNow()

6.自己实现线程池

  1. import java.util.concurrent.*;
  2.  
  3. // 线程池类
  4. public class MyThreadPoolExecutor implements Executor {
  5.     // 创建线程的工厂对象
  6.     private final ThreadFactory threadFactory;
  7.  
  8.     // 临时工摸鱼的时间上限
  9.     private final long keepAliveTime;
  10.     private final TimeUnit unit;
  11.  
  12.     // 当前正式员工的数量
  13.     private int currentCoreSize;
  14.  
  15.     // 正式员工的数量上限
  16.     private final int corePoolSize;
  17.  
  18.     // 当前临时员工的数量
  19.     private int currentTemporarySize;
  20.  
  21.     // 临时员工的数量上限
  22.     private final int temporaryPoolSize;
  23.  
  24.     // 传递任务的阻塞队列
  25.     private final BlockingQueue<Runnable> workQueue;
  26.  
  27.     public MyThreadPoolExecutor(int corePoolSize,
  28.                                 int maximumPoolSize,
  29.                                 long keepAliveTime,
  30.                                 TimeUnit unit,
  31.                                 BlockingQueue<Runnable> workQueue,
  32.                                 ThreadFactory threadFactory,
  33.                                 RejectedExecutionHandler handler) {
  34.         this.corePoolSize = corePoolSize;
  35.         this.temporaryPoolSize = maximumPoolSize - corePoolSize;
  36.         this.workQueue = workQueue;
  37.         this.threadFactory = threadFactory;
  38.         this.keepAliveTime = keepAliveTime;
  39.         this.unit = unit;
  40.     }
  41.  
  42.     // 向线程池中提交任务
  43.     // 目前这个方法不是线程安全版本的
  44.     @Override
  45.     public void execute(Runnable command) {
  46.         // 1. 如果正式员工的数量还低于正式员工的数量上限，则优先创建正式员工处理任务
  47.         // 1.1 需要管理，当前正式员工有多少，正式员工的数量上限有多少？
  48.         if (currentCoreSize < corePoolSize) {
  49.             // 优先创建正式员工进行处理
  50.             // 创建一个线程，这个线程中的任务就是不断地取任务-做任务，但是不需要考虑退出的问题
  51.             CoreJob job = new CoreJob(workQueue, command);
  52. //            Thread thread = new Thread(job);    // 不使用工厂创建的线程
  53.             Thread thread = threadFactory.newThread(job);   // thread 代表的就是正式员工
  54.             String name = String.format("正式员工-%d", currentCoreSize);
  55.             thread.setName(name);
  56.  
  57.             thread.start();
  58.  
  59.             // 只是两种不同的策略，没有谁是正确的说法
  60.             // 1. 把 command 放到队列中；command 的执行次序是在队列已有的任务之后
  61.             // 2. 创建正式员工的时候，就把 command 提交给正式员工，让 command 优先执行
  62.             // 我们这里采用第二种方案，主要原因就是 java 官方的就是使用的第二种策略
  63.  
  64.             currentCoreSize++;
  65.             return;
  66.         }
  67.  
  68.         // 走到这里，说明正式员工的数量 == 正式员工的上限了
  69.         // 2. 优先把任务放入队列中，如果放入成功，execute 执行结束，否则还需要继续
  70.         // 2.1 需要一个阻塞队列
  71. //        workQueue.put(command); // 带阻塞的放入，是否满足这里的需求？
  72.                                 // 我们这里希望的是立即得到结果
  73.         boolean success = workQueue.offer(command);
  74.         if (success == true) {
  75.             // 说明放入队列成功
  76.             return;
  77.         }
  78.  
  79.         // 队列也已经放满了
  80.         // 3. 继续判断，临时工的数量有没有到上限，如果没有到达，创建新的临时工来处理
  81.         if (currentTemporarySize < temporaryPoolSize) {
  82.             // 创建临时工进行处理
  83.             TemporaryJob job = new TemporaryJob(keepAliveTime, unit, workQueue, command);
  84. //            Thread thread = new Thread(job);    // 不使用工厂创建的线程
  85.             Thread thread = threadFactory.newThread(job);   // thread 代表的就是临时员工
  86.             String name = String.format("临时员工-%d", currentTemporarySize);
  87.             thread.setName(name);
  88.  
  89.             thread.start();
  90.  
  91.             currentTemporarySize++;
  92.             return;
  93.         }
  94.  
  95.         // 4. 执行拒绝策略
  96.         // 为了实现方便，暂时不考虑其他策略
  97.         throw new RejectedExecutionException();
  98.     }
  99. }
  1. import java.util.concurrent.BlockingQueue;
  2.  
  3. // 一个正式员工线程要完成的工作
  4. public class CoreJob implements Runnable {
  5.     // 需要阻塞队列
  6.     private final BlockingQueue<Runnable> workQueue;
  7.     private Runnable firstCommand;
  8.  
  9.     CoreJob(BlockingQueue<Runnable> workQueue, Runnable firstCommand) {
  10.         this.workQueue = workQueue;
  11.         this.firstCommand = firstCommand;
  12.     }
  13.  
  14.     @Override
  15.     public void run() {
  16.         try {
  17.             firstCommand.run();     // 优先先把刚提交的任务先做掉了
  18.             firstCommand = null;    // 这里设置 null 的意思是，不影响 firstCommand 对象被 GC 时的回收
  19.  
  20.             while (!Thread.interrupted()) {
  21.                 Runnable command = workQueue.take();
  22.                 command.run();
  23.             }
  24.         } catch (InterruptedException ignored) {}
  25.     }
  26. }
  1. import java.util.concurrent.BlockingQueue;
  2. import java.util.concurrent.TimeUnit;
  3.  
  4. // 一个临时员工线程要完成的工作
  5. public class TemporaryJob implements Runnable {
  6.     // 需要阻塞队列
  7.     private final BlockingQueue<Runnable> workQueue;
  8.     private final long keepAliveTime;
  9.     private final TimeUnit unit;
  10.     private Runnable firstCommand;
  11.  
  12.     TemporaryJob(long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Runnable firstCommand) {
  13.         this.keepAliveTime = keepAliveTime;
  14.         this.unit = unit;
  15.         this.workQueue = workQueue;
  16.         this.firstCommand = firstCommand;
  17.     }
  18.  
  19.     @Override
  20.     public void run() {
  21.         try {
  22.             firstCommand.run();     // 优先先把刚提交的任务先做掉了
  23.             firstCommand = null;    // 这里设置 null 的意思是，不影响 firstCommand 对象被 GC 时的回收
  24.  
  25.             // 一旦超过一定时间没有任务，临时工是需要退出的
  26.             // 1. keepAliveTime + unit 记录起来
  27.             // 2. 怎么就知道超过多久没有任务了？如果一定时间内都无法从队列中取出来任务，则认为摸鱼时间够了
  28.             while (!Thread.interrupted()) {
  29. //                Runnable command = workQueue.take();
  30.                 Runnable command = workQueue.poll(keepAliveTime, unit);
  31.                 if (command == null) {
  32.                     // 说明，没有取到任务
  33.                     // 说明超时时间已到
  34.                     // 说明该线程已经 keepAliveTime + unit 时间没有工作了
  35.                     // 所以，可以退出了
  36.                     break;
  37.                 }
  38.                 command.run();
  39.             }
  40.         } catch (InterruptedException ignored) {}
  41.     }
  42. }
  1. import java.util.concurrent.*;
  2.  
  3. public class Main {
  4.     static class Task implements Runnable {
  5.         @Override
  6.         public void run() {
  7.             try {
  8.                 TimeUnit.SECONDS.sleep(15);
  9.             } catch (InterruptedException ignored) {}
  10.         }
  11.     }
  12.  
  13.     static class MyThreadFactory implements ThreadFactory {
  14.         @Override
  15.         public Thread newThread(Runnable r) {
  16.             return new Thread(r);
  17.         }
  18.     }
  19.  
  20.     public static void main(String[] args) {
  21.         BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(5);
  22.         // 同时最多有 15 个任务
  23.         // 3 个正式的
  24.         // 5 个队列中
  25.         // 7 个临时的
  26.         // 提交第 16 个任务时就会出现拒绝服务
  27.         MyThreadPoolExecutor executor = new MyThreadPoolExecutor(
  28.                 3, 10, 10, TimeUnit.SECONDS,
  29.                 workQueue,
  30.                 new MyThreadFactory(),
  31.                 new ThreadPoolExecutor.AbortPolicy()
  32.         );
  33.  
  34.         // [0, 1, 2] 交给正式员工在处理
  35.         // [3, 4, 5, 6, 7] 暂时放在队列中
  36.         // [8, 9, 10, 11, 12, 13, 14] 交给临时工处理
  37.         // 过了 15s 之后，第一批任务执行结束
  38.         // [0, 1, 2]、[8, 9, 10, 11, 12, 13, 14] 执行结束
  39.         // 剩下的 [3, 4, 5, 6, 7] 任务具体怎么分配不确定，大概率是交给正式员工执行
  40.         // 就算极端情况下，5 个全部给了临时工
  41.         // 也至少还有 2 个临时工没有工作
  42.         // 再过 10s，至少 2 个，最多 5 个临时工要被解雇
  43.         for (int i = 0; i < 15; i++) {
  44.             System.out.println("提交任务: " + i);
  45.             Task task = new Task();
  46.             executor.execute(task);
  47.         }
  48.     }
  49. }

线程池总结：（⭐）

  1. 为什么
  2. 方法 + 机制
    • 构造方法的含义
    • execute的使用
    • 其他的使用暂时没有介绍
  3. 自己实现（难点）
声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号