自定义线程池（二）

上节回顾

在上一节当中，已经实现了一个线程池，在本节当中，我们需要添加拒绝策略。这里使用到了策略模式的设计模式，由于拒绝策略是多种的，我们需要将这个权利下放给调用者（由调用者来指定我要采取哪种策略），而线程池只需要调用拒绝的接口即可。

步骤

（1）定义拒绝策略接口
（2）在线程池中加入拒绝策略参数
（3）自行调用测试
1.定义接口类

1. @FunctionalInterface
2. interface RejectPolicy<T>{
3.     //注意传递参数
4.     void reject(BlockQueue<T> queue,Runnable task);
5. }

复制代码

2.线程池中添加接口以及调用方法

1. @Slf4j
2. class ThreadPool {
3.     //任务队列
4.     private BlockQueue<Runnable> taskQueue;
5.     //线程集合 我们需要对线程做一个包装
6.     private HashSet<Worker> workers = new HashSet<>();
7.     //核心线程数量
8.     private long coreSize;
9.     //超时时间
10.     private long timeout;
11.     //时间单位
12.     private TimeUnit timeUnit;
13.     //自定义拒绝策略
14.     private RejectPolicy<Runnable> rejectPolicy;
16.     public ThreadPool(int queueCapacity,long coreSize,long timeout,TimeUnit timeUnit){
17.         this.taskQueue = new BlockQueue<>(queueCapacity);
18.         this.coreSize = coreSize;
19.         this.timeout = timeout;
20.         this.timeUnit = timeUnit;
21.         this.rejectPolicy = (queue, task) -> {
22.             throw new RuntimeException();
23.         };
24.     }
25.     public ThreadPool(int queueCapacity,long coreSize,long timeout,TimeUnit timeUnit,RejectPolicy<Runnable> rejectPolicy){
26.         taskQueue = new BlockQueue<>(queueCapacity);
27.         this.coreSize = coreSize;
28.         this.timeout = timeout;
29.         this.timeUnit = timeUnit;
30.         this.rejectPolicy = rejectPolicy;
31.     }
32.     //执行任务
33.     public void execute(Runnable task){
34.         //当任务数量尚未超过coreSize
35.         synchronized (workers){
36.             if (workers.size() < coreSize){
37.                 log.info("创建工作线程{}",task);
38.                 Worker worker = new Worker(task);
39.                 workers.add(worker);
40.                 worker.start();
41.             }else{
42.                 log.info("加入到任务队列{}",task);
43.                 //有可能会阻塞在这里 进而将主线程阻塞掉
44.                 //taskQueue.put(task);
45.                 //这里会有很多种策略自定义策略
46.                 //策略模式：操作抽象成接口实现代码是传过来不会写死
47.                 taskQueue.tryPut(rejectPolicy,task);
48.                 //rejectPolicy.reject(taskQueue,task);
49.             }
50.         }
51.     }
53.     class Worker extends Thread{
54.         private Runnable task;
55.         public Worker(Runnable task){
56.             this.task = task;
57.         }
59.         @Override
60.         public void run() {
61.            while (task != null || (task = taskQueue.poll(timeout,timeUnit)) != null){
62.                try {
63.                    log.info("正在执行...{}",task);
64.                    //执行任务
65.                    task.run();
66.                }catch (Exception e){
67.                    System.out.println(e.getMessage());
68.                }finally {
69.                    //不要忘记这一步
70.                    task = null;
71.                }
72.            }
73.            synchronized (workers){
74.                log.info("worker被移除{}",this);
75.                workers.remove(this);
76.            }
77.         }
78.     }
79. }

复制代码

3.main测试

1. @Slf4j
2. public class TestPool {
3.     //阻塞队列是平衡生产者和消费者之间的中介
4.     //任务数量超过任务队列的情况
5.     public static void main(String[] args) {
6.         ThreadPool threadPool = new ThreadPool(10, 2, 1000, TimeUnit.MICROSECONDS, (queue, task) -> {
7.             //1.死等
8.             queue.put(task);
9.             //2.超时等待
10.             queue.offer(task, 1500, TimeUnit.MICROSECONDS);
11.             //3.调用者自己放弃
12.             // log.debug("放弃{}",task);
13.             //4.调用者抛异常
14.             //throw new RuntimeException("task执行失败" + task);
15.             //5.调用者自己执行
16.             task.run();
17.         });
18.         for (int i = 0; i < 20; i++) {
19.             int j = i;
20.             //主线程可能会在这里阻塞
21.             threadPool.execute(() -> {
22.                 try {
23.                     Thread.sleep(30000);
24.                 } catch (InterruptedException e) {
25.                     throw new RuntimeException(e);
26.                 }
27.                 TestPool.log.debug("{}", j);
28.             });
29.         }
30.     }
31. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。