1. package com.lms.jdk8.juc;
    3. import lombok.extern.slf4j.Slf4j;
    5. import java.util.ArrayDeque;
    6. import java.util.Deque;
    7. import java.util.HashSet;
    8. import java.util.concurrent.TimeUnit;
    9. import java.util.concurrent.locks.Condition;
    10. import java.util.concurrent.locks.ReentrantLock;
    12. /**
    13.  * @Author: 李孟帅
    14.  * @Date: 2021-12-16 10:33
    15.  * @Description:
    16.  */
    17. @Slf4j
    18. public class ThreadPool {
    19.     // 任务队列
    20.     private BlockQueue<Runnable> taskQueue;
    21.     // 线程集合
    22.     private final HashSet<Worker> workers = new HashSet<>();
    23.     // 核心线程数
    24.     private int coreSize;
    25.     // 获取任务的超时时间
    26.     private long timeout;
    27.     private TimeUnit timeUnit;
    28.     // 拒绝策略
    29.     private RejectPolicy<Runnable> rejectPolicy;
    31.     public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, BlockQueue<Runnable> taskQueue, RejectPolicy<Runnable> rejectPolicy) {
    32.         this.coreSize = coreSize;
    33.         this.timeout = timeout;
    34.         this.timeUnit = timeUnit;
    35.         this.taskQueue = taskQueue;
    36.         this.rejectPolicy = rejectPolicy;
    37.     }
    39.     // 执行任务
    40.     public void execute(Runnable task) {
    41.         synchronized (workers) {
    42.             // 任务数没有超过coreSize时，直接交给worker执行，否则加入队列暂存
    43.             if (workers.size() < coreSize) {
    44.                 Worker worker = new Worker(task);
    45.                 log.info("创建worker {},task:{}", worker, task);
    46.                 workers.add(worker);
    47.                 worker.start();
    48.             } else {
    49.                 // 交给拒绝策略执行
    50.                 taskQueue.tryPut(rejectPolicy, task);
    51.             }
    52.         }
    53.     }
    55.     class Worker extends Thread {
    57.         private Runnable task;
    59.         public Worker(Runnable task) {
    60.             this.task = task;
    61.         }
    63.         @Override
    64.         public void run() {
    66.             // 执行任务
    67.             // task不为null，执行任务
    68.             // task为null,在从任务队列中获取任务并执行
    69.             while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
    70.                 try {
    71.                     log.info("正在执行...{}", task);
    72.                     task.run();
    73.                 } catch (Exception e) {
    74.                     e.printStackTrace();
    75.                 } finally {
    76.                     task = null;
    77.                 }
    78.             }
    79.             synchronized (workers) {
    80.                 log.debug("worker 被移除{}", this);
    81.                 workers.remove(this);
    82.             }
    83.         }
    84.     }
    85. }
    87. @Slf4j
    88. class BlockQueue<T> {
    89.     // 存放任务队列
    90.     private Deque<T> queue = new ArrayDeque<>();
    91.     // 容量
    92.     private int capacity;
    93.     // 锁
    94.     private ReentrantLock lock = new ReentrantLock();
    95.     // 生产者条件变量
    96.     private Condition fullCWait = lock.newCondition();
    97.     // 消费者条件变量
    98.     private Condition emptyWait = lock.newCondition();
    100.     public BlockQueue(int capacity) {
    101.         this.capacity = capacity;
    102.     }
    104.     // 带超时阻塞获取
    105.     public T poll(long timeout, TimeUnit unit) {
    106.         lock.lock();
    107.         try {
    108.             long nanos = unit.toNanos(timeout);
    109.             while (queue.isEmpty()) {
    110.                 try {
    111.                     if (nanos <= 0) {
    112.                         return null;
    113.                     }
    114.                     // 返回值是剩余时间
    115.                     nanos = emptyWait.awaitNanos(nanos);
    116.                 } catch (InterruptedException e) {
    117.                     e.printStackTrace();
    118.                 }
    119.             }
    120.             T t = queue.removeFirst();
    121.             fullCWait.signal();
    122.             return t;
    123.         } finally {
    124.             lock.unlock();
    125.         }
    126.     }
    128.     // 阻塞获取
    129.     public T take() {
    130.         lock.lock();
    131.         try {
    132.             while (queue.isEmpty()) {
    133.                 try {
    134.                     emptyWait.await();
    135.                 } catch (InterruptedException e) {
    136.                     e.printStackTrace();
    137.                 }
    138.             }
    139.             T t = queue.removeFirst();
    140.             fullCWait.signal();
    141.             return t;
    142.         } finally {
    143.             lock.unlock();
    144.         }
    145.     }
    147.     // 阻塞添加
    148.     public void put(T element) {
    149.         lock.lock();
    150.         try {
    151.             while (queue.size() == capacity) {
    152.                 try {
    153.                     log.info("等待加入任务队列 {}...", element);
    154.                     fullCWait.await();
    155.                 } catch (InterruptedException e) {
    156.                     e.printStackTrace();
    157.                 }
    158.             }
    159.             log.info("加入任务队列 {}", element);
    160.             queue.addLast(element);
    161.             emptyWait.signal();
    162.         } finally {
    163.             lock.unlock();
    164.         }
    165.     }
    167.     // 带超时时间阻塞添加
    168.     public boolean offer(T element, long timeout, TimeUnit unit) {
    169.         lock.lock();
    170.         try {
    171.             long nanos = unit.toNanos(timeout);
    172.             while (queue.size() == capacity) {
    173.                 try {
    174.                     if (nanos <= 0) {
    175.                         log.info("等待加入任务队列超时 {}...", element);
    176.                         return false;
    177.                     }
    178.                     nanos = fullCWait.awaitNanos(nanos);
    179.                 } catch (InterruptedException e) {
    180.                     e.printStackTrace();
    181.                 }
    182.             }
    183.             log.info("加入任务队列 {}", element);
    184.             queue.addLast(element);
    185.             emptyWait.signal();
    186.             return true;
    187.         } finally {
    188.             lock.unlock();
    189.         }
    190.     }
    192.     // 队列长度
    193.     public int size() {
    194.         lock.lock();
    195.         try {
    196.             return queue.size();
    197.         } finally {
    198.             lock.unlock();
    199.         }
    200.     }
    202.     // 尝试加入队列，失败时 拒绝策略
    203.     public void tryPut(RejectPolicy<T> rejectPolicy, T element) {
    204.         lock.lock();
    205.         try {
    206.             // 队列是否已满
    207.             if (queue.size() == capacity) {
    208.                 rejectPolicy.reject(this, element);
    209.             } else { //有空闲 加入队列
    210.                 log.info("加入任务队列 {}", element);
    211.                 queue.addLast(element);
    212.                 emptyWait.signal();
    213.             }
    214.         } finally {
    215.             lock.unlock();
    216.         }
    218.     }
    219. }
    221. @FunctionalInterface
    222. interface RejectPolicy<T> {
    223.     void reject(BlockQueue<T> queue, T element);
    224. }
    226. @Slf4j
    227. class Test {
    228.     public static void main(String[] args) {
    229.         ThreadPool threadPool = new ThreadPool(2, 1000, TimeUnit.MILLISECONDS, new BlockQueue<>(1), (queue, element) -> {
    230.             // 死等
    231.             // queue.put(element);
    232.             // 超时等待
    233.             // queue.offer(element,500,TimeUnit.MILLISECONDS);
    234.             // 让调用者自己放弃
    235.             // log.info("放弃{}", element);
    236.             // 让调用者抛出异常
    237.             // throw new RuntimeException("任务执行失败："+element);
    238.             // 让调用者自己执行任务
    239.             element.run();
    240.         });
    242.         for (int i = 0; i < 5; i++) {
    243.             int finalI = i;
    244.             threadPool.execute(() -> {
    245.                 try {
    246.                     Thread.sleep(1000);
    247.                     log.info("run :{}", finalI);
    248.                 } catch (InterruptedException e) {
    249.                     e.printStackTrace();
    250.                 }
    251.             });
    253.         }
    254.     }
    255. }