• Java并发-实现带拒绝策略的线程池

    目录

    1 前言

    2 线程池

     2.1 线程池的设计

    2.2 阻塞队列

    2.3 线程池实现

    2.4 拒绝策略

    2.5 完整代码

    1 前言

    Java多线程的学习是绕不开线程池的，线程池可以帮助我们管理线程，处理任务。在学习多线程时我已经实现过一个简单的线程池《Java并发-线程池的实现》这个这是简单的阻塞等待。

    2 线程池

    如下图所示：线程池最根本的应该是生产者消费者模式，生产者向线程池中提供任务，线程池中的线程进行任务消费。

     2.1 线程池的设计

    在本节我们将设计一个线程池

    • 阻塞队列
    • 线程池
    • 拒绝策略
      • 阻塞等待
      • 超时等待
      • 放弃任务
      • 报错
      • 调用者线程自己执行

    2.2 阻塞队列

    属性

    • 队列容量
    • 真实队列
    • 锁ReentrantLock
    • 生产者的Condition
    • 消费者的Condition

    take方法

    • 阻塞实现
    • 超时等待实现：是对阻塞等待的改进，可以在没有任务的时候将Worker移除，防止线程一直阻塞，有任务时候在进行创建。

    put方法

    • 阻塞实现
    • 超时等待实现

    tryPut方法：用于拒绝策略

    1. {
    2.     // 1. 队列的长度
    3.     private int capacity;
    4.  
    5.     // 2.任务队列
    6.     private volatile Deque queue = new ArrayDeque<>();
    7.  
    8.     // 3. 锁
    9.     private ReentrantLock lock = new ReentrantLock();
    10.  
    11.     // 4.阻塞条件
    12.     // 4.1 队列为满时，生产者进入
    13.     private Condition fullWaitSet = lock.newCondition();
    14.     // 4.1 队列为空时，消费者进入
    15.     private Condition emptyWaitSet = lock.newCondition();
    16.  
    17.     public BlockQueue(int capacity) {
    18.         this.capacity = capacity;
    19.     }
    20.  
    21.     // 阻塞获取
    22.     public T take() {
    23.         lock.lock(); // 每次只能有一个线程可以消费
    24.         try {
    25.             while (queue.isEmpty()) {
    26.                 try {
    27.                     emptyWaitSet.await();
    28.                 } catch (InterruptedException e) {
    29.                     e.printStackTrace();
    30.                 }
    31.             }
    32.             T t = queue.removeFirst();
    33.             fullWaitSet.signal();
    34.             return t;
    35.         } finally {
    36.             lock.unlock();
    37.         }
    38.     }
    39.  
    40.     // 超时时间的获取
    41.     public T take(long time, TimeUnit timeUnit) {
    42.         lock.lock(); // 每次只能有一个线程可以消费
    43.         try {
    44.             long nanos = timeUnit.toNanos(time);
    45.             while (queue.isEmpty()) {
    46.                 try {
    47.                     if (nanos <= 0) {
    48.                         // 等待超时---结束循环
    49.                         return null;
    50.                     }
    51.                     nanos = emptyWaitSet.awaitNanos(nanos);
    52.                 } catch (InterruptedException e) {
    53.                     e.printStackTrace();
    54.                 }
    55.             }
    56.             T t = queue.removeFirst();
    57.             fullWaitSet.signal();
    58.             return t;
    59.         } finally {
    60.             lock.unlock();
    61.         }
    62.     }
    63.  
    64.     // 阻塞 添加
    65.     public void put(T task) {
    66.         lock.lock();
    67.         try {
    68.             while (queue.size() == capacity) {
    69.                 try {
    70.                     System.out.println("等待加入任务对列" + task);
    71.                     fullWaitSet.await();
    72.                 } catch (InterruptedException e) {
    73.                     e.printStackTrace();
    74.                 }
    75.             }
    76.             queue.addLast(task);
    77.             System.out.println("加入任务队列" + task);
    78.             emptyWaitSet.signal();
    79.         } finally {
    80.             lock.unlock();
    81.         }
    82.     }
    83.  
    84.     // 超时 添加
    85.     public boolean put(T task, long time, TimeUnit unit) {
    86.         lock.lock();
    87.         try {
    88.             long nanos = unit.toNanos(time);
    89.             while (queue.size() == capacity) {
    90.                 try {
    91.                     if (nanos <= 0) {
    92.                         return false;
    93.                     }
    94.                     System.out.println("等待加入任务队列 ..." + task);
    95.                     nanos = fullWaitSet.awaitNanos(nanos);
    96.                 } catch (InterruptedException e) {
    97.                     e.printStackTrace();
    98.                 }
    99.             }
    100.             queue.addLast(task);
    101.             System.out.println("加入任务队列 ..." + task);
    102.             emptyWaitSet.signal();
    103.             return true;
    104.         } finally {
    105.             lock.unlock();
    106.         }
    107.     }
    108.  
    109.     // 获取任务的大小。
    110.     public int size() {
    111.         lock.lock();
    112.         try {
    113.             return queue.size();
    114.         } finally {
    115.             lock.unlock();
    116.         }
    117.     }
    118.  
    119.     public void tryput(RejectPolicy rejectPolicy, T task) {
    120.         lock.lock();
    121.         try {
    122.             // 1. 队列是否满了
    123.             if (queue.size() == capacity) {
    124.                 // 2. 实行拒绝策略
    125.                 rejectPolicy.reject(this,task);
    126.             }else {
    127.                 // 有空闲
    128.                 queue.addLast(task);
    129.                 System.out.println("加入任务队列 ..." + task);
    130.                 emptyWaitSet.signal();
    131.             }
    132.         }finally {
    133.             lock.unlock();
    134.         }
    135.     }
    136. }

    2.3 线程池实现

    属性

    • 阻塞队列
    • 超时时间
    • 时间单位
    • 工作线程集合
    • 拒绝策略
    • 工作线程
    1. class ThreadPool {
    2.     private final BlockQueue taskQueue;
    3.  
    4.     private final int coreSize;
    5.  
    6.     private long time;
    7.  
    8.     private TimeUnit timeUnit;
    9.  
    10.     private final HashSet workers = new HashSet<>();
    11.  
    12.     private RejectPolicy rejectPolicy;
    13.  
    14.     public ThreadPool(int coreSize, long time, TimeUnit timeUnit, int queueCapacity ,RejectPolicy rejectPolicy) {
    15.         this.coreSize = coreSize;
    16.         this.time = time;
    17.         this.timeUnit = timeUnit;
    18.         this.taskQueue = new BlockQueue<>(queueCapacity);
    19.         this.rejectPolicy = rejectPolicy;
    20.     }
    21.  
    22.     public void execute(Runnable task) {
    23.         // 当任务数没有超过核心线程数时直接执行。
    24.         // 当任务数超过核心线程数时候，选择拒绝策略进行执行
    25.         synchronized (workers) {
    26.             if (workers.size() < coreSize) {
    27.                 Worker worker = new Worker(task);
    28.                 System.out.println("新增 work" + worker);
    29.                 workers.add(worker);
    30.                 worker.start();
    31.             } else {
    32.                 // 拒绝策略
    33.                 taskQueue.tryput(rejectPolicy,task);
    34.  
    35.             }
    36.         }
    37.     }
    38.  
    39.     // 真正工作的线程
    40.     class Worker extends Thread {
    41.         private Runnable task;
    42.  
    43.         public Worker(Runnable task) {
    44.             this.task = task;
    45.         }
    46.  
    47.         @Override
    48.         public void run() {
    49.             // 当task不为空，执行任务
    50.             // 当task执行完毕，再接着任务队列获取任务并执行---take会一直等待有没有新的任务。因此在没有任务时是阻塞的
    51.             while (task != null || (task = taskQueue.take(time, timeUnit)) != null) {
    52.                 try {
    53.                     System.out.println("正在执行..." + task);
    54.                     task.run();
    55.                 } catch (Exception e) {
    56.                     e.printStackTrace();
    57.                 } finally {
    58.                     task = null;
    59.                 }
    60.             }
    61.             synchronized (workers) {
    62.                 System.out.println("当前worker使用完 被移除 " + this);
    63.                 workers.remove(this);
    64.             }
    65.         }
    66.  
    67.         @Override
    68.         public boolean equals(Object o) {
    69.             if (this == o) {
    70.                 return true;
    71.             }
    72.             if (o == null || getClass() != o.getClass()) {
    73.                 return false;
    74.             }
    75.  
    76.             Worker worker = (Worker) o;
    77.  
    78.             return task != null ? task.equals(worker.task) : worker.task == null;
    79.         }
    80.  
    81.         @Override
    82.         public int hashCode() {
    83.             return task != null ? task.hashCode() : 0;
    84.         }
    85.     }
    86.  
    87.     public int size() {
    88.         return taskQueue.size();
    89.     }
    90. }

    2.4 拒绝策略

    拒绝策略是一个接口，作为线程池的一个参数，在线程初始化时候确定。

    1. // 拒绝策略
    2. @FunctionalInterface
    3. interface RejectPolicy {
    4.     void reject(BlockQueue queue, T task);
    5. }

    2.5 完整代码

    1. package cn.itcast.n6.c5;
    2.  
    3. import java.util.ArrayDeque;
    4. import java.util.Deque;
    5. import java.util.HashSet;
    6. import java.util.concurrent.TimeUnit;
    7. import java.util.concurrent.locks.Condition;
    8. import java.util.concurrent.locks.ReentrantLock;
    9.  
    10. /**
    11.  * @author : msf
    12.  * @date : 2022/12/5
    13.  */
    14. public class MyThreadPool {
    15.  
    16.     public static void main(String[] args) {
    17.         ThreadPool execuror = new ThreadPool(
    18.                 1, 1, TimeUnit.SECONDS, 1,
    19.                 (queue, task) -> {
    20.                     // 1) 死等
    21.                     //queue.put(task);
    22.                     // 2) 带超时等待
    23.                     //queue.put(task,1500,TimeUnit.MILLISECONDS);
    24.                     // 3) 让调用者放弃任务执行 --- 什么都不写。任务没有加在队列里面，也没有尝试等待
    25.                     // 4) 让调用者抛出异常---剩余的任务都不在执行了。
    26.                         //throw new RuntimeException("任务执行失败" + task);
    27.                     // 5) 让调用者自己执行任
    28.                     task.run();
    29.                 });
    30.  
    31.  
    32.         for (int i = 0; i < 3; i++) {
    33.             int j = i;
    34.             execuror.execute(() -> {
    35.                 try {
    36.                     Thread.sleep(500);
    37.                 } catch (InterruptedException e) {
    38.                     e.printStackTrace();
    39.                 }
    40.                 System.out.println(Thread.currentThread().getName() + "执行了" + j);
    41. //                System.out.println(execuror.size());
    42.             });
    43.         }
    44.     }
    45. }
    46.  
    47. // 拒绝策略
    48. @FunctionalInterface
    49. interface RejectPolicy {
    50.     void reject(BlockQueue queue, T task);
    51. }
    52.  
    53.  
    54. class ThreadPool {
    55.     private final BlockQueue taskQueue;
    56.  
    57.     private final int coreSize;
    58.  
    59.     private long time;
    60.  
    61.     private TimeUnit timeUnit;
    62.  
    63.     private final HashSet workers = new HashSet<>();
    64.  
    65.     private RejectPolicy rejectPolicy;
    66.  
    67.     public ThreadPool(int coreSize, long time, TimeUnit timeUnit, int queueCapacity ,RejectPolicy rejectPolicy) {
    68.         this.coreSize = coreSize;
    69.         this.time = time;
    70.         this.timeUnit = timeUnit;
    71.         this.taskQueue = new BlockQueue<>(queueCapacity);
    72.         this.rejectPolicy = rejectPolicy;
    73.     }
    74.  
    75.     public void execute(Runnable task) {
    76.         // 当任务数没有超过核心线程数时直接执行。
    77.         // 当任务数超过核心线程数时候，选择拒绝策略进行执行
    78.         synchronized (workers) {
    79.             if (workers.size() < coreSize) {
    80.                 Worker worker = new Worker(task);
    81.                 System.out.println("新增 work" + worker);
    82.                 workers.add(worker);
    83.                 worker.start();
    84.             } else {
    85.                 // 拒绝策略
    86.                 taskQueue.tryput(rejectPolicy,task);
    87.  
    88.             }
    89.         }
    90.     }
    91.  
    92.     // 真正工作的线程
    93.     class Worker extends Thread {
    94.         private Runnable task;
    95.  
    96.         public Worker(Runnable task) {
    97.             this.task = task;
    98.         }
    99.  
    100.         @Override
    101.         public void run() {
    102.             // 当task不为空，执行任务
    103.             // 当task执行完毕，再接着任务队列获取任务并执行---take会一直等待有没有新的任务。因此在没有任务时是阻塞的
    104.             while (task != null || (task = taskQueue.take(time, timeUnit)) != null) {
    105.                 try {
    106.                     System.out.println("正在执行..." + task);
    107.                     task.run();
    108.                 } catch (Exception e) {
    109.                     e.printStackTrace();
    110.                 } finally {
    111.                     task = null;
    112.                 }
    113.             }
    114.             synchronized (workers) {
    115.                 System.out.println("当前worker使用完 被移除 " + this);
    116.                 workers.remove(this);
    117.             }
    118.         }
    119.  
    120.         @Override
    121.         public boolean equals(Object o) {
    122.             if (this == o) {
    123.                 return true;
    124.             }
    125.             if (o == null || getClass() != o.getClass()) {
    126.                 return false;
    127.             }
    128.  
    129.             Worker worker = (Worker) o;
    130.  
    131.             return task != null ? task.equals(worker.task) : worker.task == null;
    132.         }
    133.  
    134.         @Override
    135.         public int hashCode() {
    136.             return task != null ? task.hashCode() : 0;
    137.         }
    138.     }
    139.  
    140.     public int size() {
    141.         return taskQueue.size();
    142.     }
    143. }
    144.  
    145. class BlockQueue {
    146.     // 1. 队列的长度
    147.     private int capacity;
    148.  
    149.     // 2.任务队列
    150.     private volatile Deque queue = new ArrayDeque<>();
    151.  
    152.     // 3. 锁
    153.     private ReentrantLock lock = new ReentrantLock();
    154.  
    155.     // 4.阻塞条件
    156.     // 4.1 队列为满时，生产者进入
    157.     private Condition fullWaitSet = lock.newCondition();
    158.     // 4.1 队列为空时，消费者进入
    159.     private Condition emptyWaitSet = lock.newCondition();
    160.  
    161.     public BlockQueue(int capacity) {
    162.         this.capacity = capacity;
    163.     }
    164.  
    165.     // 阻塞获取
    166.     public T take() {
    167.         lock.lock(); // 每次只能有一个线程可以消费
    168.         try {
    169.             while (queue.isEmpty()) {
    170.                 try {
    171.                     emptyWaitSet.await();
    172.                 } catch (InterruptedException e) {
    173.                     e.printStackTrace();
    174.                 }
    175.             }
    176.             T t = queue.removeFirst();
    177.             fullWaitSet.signal();
    178.             return t;
    179.         } finally {
    180.             lock.unlock();
    181.         }
    182.     }
    183.  
    184.     // 超时时间的获取
    185.     public T take(long time, TimeUnit timeUnit) {
    186.         lock.lock(); // 每次只能有一个线程可以消费
    187.         try {
    188.             long nanos = timeUnit.toNanos(time);
    189.             while (queue.isEmpty()) {
    190.                 try {
    191.                     if (nanos <= 0) {
    192.                         // 等待超时---结束循环
    193.                         return null;
    194.                     }
    195.                     nanos = emptyWaitSet.awaitNanos(nanos);
    196.                 } catch (InterruptedException e) {
    197.                     e.printStackTrace();
    198.                 }
    199.             }
    200.             T t = queue.removeFirst();
    201.             fullWaitSet.signal();
    202.             return t;
    203.         } finally {
    204.             lock.unlock();
    205.         }
    206.     }
    207.  
    208.     // 阻塞 添加
    209.     public void put(T task) {
    210.         lock.lock();
    211.         try {
    212.             while (queue.size() == capacity) {
    213.                 try {
    214.                     System.out.println("等待加入任务对列" + task);
    215.                     fullWaitSet.await();
    216.                 } catch (InterruptedException e) {
    217.                     e.printStackTrace();
    218.                 }
    219.             }
    220.             queue.addLast(task);
    221.             System.out.println("加入任务队列" + task);
    222.             emptyWaitSet.signal();
    223.         } finally {
    224.             lock.unlock();
    225.         }
    226.     }
    227.  
    228.     // 超时 添加
    229.     public boolean put(T task, long time, TimeUnit unit) {
    230.         lock.lock();
    231.         try {
    232.             long nanos = unit.toNanos(time);
    233.             while (queue.size() == capacity) {
    234.                 try {
    235.                     if (nanos <= 0) {
    236.                         return false;
    237.                     }
    238.                     System.out.println("等待加入任务队列 ..." + task);
    239.                     nanos = fullWaitSet.awaitNanos(nanos);
    240.                 } catch (InterruptedException e) {
    241.                     e.printStackTrace();
    242.                 }
    243.             }
    244.             queue.addLast(task);
    245.             System.out.println("加入任务队列 ..." + task);
    246.             emptyWaitSet.signal();
    247.             return true;
    248.         } finally {
    249.             lock.unlock();
    250.         }
    251.     }
    252.  
    253.     // 获取任务的大小。
    254.     public int size() {
    255.         lock.lock();
    256.         try {
    257.             return queue.size();
    258.         } finally {
    259.             lock.unlock();
    260.         }
    261.     }
    262.  
    263.     public void tryput(RejectPolicy rejectPolicy, T task) {
    264.         lock.lock();
    265.         try {
    266.             // 1. 队列是否满了
    267.             if (queue.size() == capacity) {
    268.                 // 2. 实行拒绝策略
    269.                 rejectPolicy.reject(this,task);
    270.             }else {
    271.                 // 有空闲
    272.                 queue.addLast(task);
    273.                 System.out.println("加入任务队列 ..." + task);
    274.                 emptyWaitSet.signal();
    275.             }
    276.         }finally {
    277.             lock.unlock();
    278.         }
    279.     }
    280. }

  • 相关阅读:
    tensorRT是怎么构建和编译一个模型的
    【Linux】冯诺依曼体系结构
    二叉树层级遍历（深度优先、广度优先算法）
    IDEA中Tomcat启动后提交表单，请求的资源[/servlet_demo2/book-add.html]不可用
    docker部署nacos集群
    xss跨站脚本攻击姿势大全
    第六篇Android--ImageView、Bitmap
    Dubbo和Spring Cloud相关
    【腾讯云 Cloud Studio 实战训练营】使用云IDEA，快速构建React完成点餐H5页面
    50个常用的Numpy函数解释，参数和使用示例
  • 原文地址：https://blog.csdn.net/abc123mma/article/details/128199245