JUC学习-线程池部分

自定义线程池

1. package com.appletree24;  
2.   
3. import java.util.ArrayDeque;  
4. import java.util.Deque;  
5. import java.util.HashSet;  
6. import java.util.concurrent.ExecutionException;  
7. import java.util.concurrent.TimeUnit;  
8. import java.util.concurrent.locks.Condition;  
9. import java.util.concurrent.locks.ReentrantLock;  
10.   
11. class Main {  
12.     public static void main(String[] args) throws ExecutionException, InterruptedException {  
13.         ThreadPool threadPool = new ThreadPool(2, 1000, TimeUnit.MICROSECONDS, 5, (queue, task) -> {  
14.             //带超时等待  
15. //            queue.offer(task,500,TimeUnit.MILLISECONDS);  
16.         });  
17.         for (int i = 0; i < 10; i++) {  
18.             int j = i;  
19.             threadPool.execute(() -> {  
20.                 try {  
21.                     Thread.sleep(1000L);  
22.                 } catch (InterruptedException e) {  
23.                     e.printStackTrace();  
24.                 }  
25.                 System.out.println(j);  
26.             });  
27.         }  
28.     }  
29. }  
30.   
31. //策略模式接口 此处使用策略模式是因为在实现拒绝策略时，有许多种拒绝的方式，这些方式如果不使用恰当的模式，就需要大量的if..else来编写  
32. //且方式数量大于4个，会造成类膨胀的问题，推荐使用混合模式  
33. //https://www.runoob.com/design-pattern/strategy-pattern.html  
34. @FunctionalInterface  
35. interface RejectPolicy<T> {  
36.     void reject(BlockingQueue<T> queue, T task);  
37. }  
38.   
39. class ThreadPool {  
40.     //任务队列  
41.     private BlockingQueue<Runnable> taskQueue;  
42.     //线程集合  
43.     private HashSet<Worker> workers = new HashSet<>();  
44.   
45.     //线程数  
46.     private int coreSize;  
47.   
48.     //超时时间  
49.     private long timeout;  
50.   
51.     private TimeUnit timeUnit;  
52.     private RejectPolicy<Runnable> rejectPolicy;  
53.   
54.     //执行任务  
55.     public void execute(Runnable task) {  
56.         //当任务数未超过核心线程数时，直接交给Worker对象执行  
57.         //如果超过，则加入阻塞任务队列，暂存起来  
58.         synchronized (workers) {  
59.             if (workers.size() < coreSize) {  
60.                 Worker worker = new Worker(task);  
61.                 workers.add(worker);  
62.                 worker.start();  
63.             } else {  
64.                 //第一种选择死等  
65. //                taskQueue.put(task);  
66.                 //第二种为超时等待  
67.                 //第三种为消费者放弃任务执行  
68.                 //第四种为主线程抛出异常  
69.                 //第五种让调用者自行执行任务  
70.                 taskQueue.tryPut(rejectPolicy, task);  
71.             }  
72.         }  
73.     }  
74.   
75.     public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity, RejectPolicy<Runnable> rejectPolicy) {  
76.         this.coreSize = coreSize;  
77.         this.timeout = timeout;  
78.         this.timeUnit = timeUnit;  
79.         this.taskQueue = new BlockingQueue<>(queueCapcity);  
80.         this.rejectPolicy = rejectPolicy;  
81.     }  
82.   
83.     class Worker extends Thread {  
84.         private Runnable task;  
85.   
86.         public Worker(Runnable task) {  
87.             this.task = task;  
88.         }  
89.   
90.         @Override  
91.         public void run() {  
92.             //执行任务  
93.             //1.当传递过来的task不为空，执行任务  
94.             //2.当task执行完毕，再接着取下一个任务并执行  
95.             while (task != null || (task = taskQueue.poll(1000, TimeUnit.MILLISECONDS)) != null) {  
96.                 try {  
97.                     task.run();  
98.                 } catch (Exception e) {  
99.                     e.printStackTrace();  
100.                 } finally {  
101.                     task = null;  
102.                 }  
103.             }  
104.             synchronized (workers) {  
105.                 workers.remove(this);  
106.             }  
107.         }  
108.     }  
109. }  
110.   
111. class BlockingQueue<T> {  
112.     //1. 任务队列  
113.     private final Deque<T> queue = new ArrayDeque<>();  
114.   
115.     //2. 锁  
116.     private final ReentrantLock lock = new ReentrantLock();  
117.   
118.     //3. 生产者条件变量  
119.     private final Condition fullWaitSet = lock.newCondition();  
120.   
121.     //4. 消费者条件变量  
122.     private final Condition emptyWaitSet = lock.newCondition();  
123.   
124.     //5. 容量上限  
125.     private int capcity;  
126.   
127.     public BlockingQueue(int capcity) {  
128.         this.capcity = capcity;  
129.     }  
130.   
131.     //带超时的等待获取  
132.     public T poll(long timeout, TimeUnit unit) {  
133.         lock.lock();  
134.         long nanos = unit.toNanos(timeout);  
135.         try {  
136.             while (queue.isEmpty()) {  
137.                 try {  
138.                     if (nanos <= 0) {  
139.                         return null;  
140.                     }  
141.                     nanos = emptyWaitSet.awaitNanos(nanos);  
142.                 } catch (InterruptedException e) {  
143.                     e.printStackTrace();  
144.                 }  
145.             }  
146.             T t = queue.removeFirst();  
147.             fullWaitSet.signal();  
148.             return t;  
149.         } finally {  
150.             lock.unlock();  
151.         }  
152.     }  
153.   
154.     //消费者拿取任务的方法  
155.     public T take() {  
156.         lock.lock();  
157.         try {  
158.             while (queue.isEmpty()) {  
159.                 try {  
160.                     emptyWaitSet.await();  
161.                 } catch (InterruptedException e) {  
162.                     e.printStackTrace();  
163.                 }  
164.             }  
165.             T t = queue.removeFirst();  
166.             fullWaitSet.signal();  
167.             return t;  
168.         } finally {  
169.             lock.unlock();  
170.         }  
171.     }  
172.   
173.     //阻塞添加  
174.     public void put(T task) {  
175.         lock.lock();  
176.         try {  
177.             while (queue.size() == capcity) {  
178.                 try {  
179.                     fullWaitSet.await();  
180.                 } catch (InterruptedException e) {  
181.                     e.printStackTrace();  
182.                 }  
183.             }  
184.             queue.offerLast(task);  
185.             //添加完后唤醒消费者等待  
186.             emptyWaitSet.signal();  
187.         } finally {  
188.             lock.unlock();  
189.         }  
190.     }  
191.   
192.     //带超时时间的阻塞添加  
193.     public boolean offer(T task, long timeout, TimeUnit unit) {  
194.         lock.lock();  
195.         try {  
196.             long nanos = unit.toNanos(timeout);  
197.             while (queue.size() == capcity) {  
198.                 try {  
199.                     if (nanos <= 0) return false;  
200.                     nanos = fullWaitSet.awaitNanos(nanos);  
201.                 } catch (InterruptedException e) {  
202.                     e.printStackTrace();  
203.                 }  
204.             }  
205.             queue.offerLast(task);  
206.             //添加完后唤醒消费者等待  
207.             emptyWaitSet.signal();  
208.             return true;  
209.         } finally {  
210.             lock.unlock();  
211.         }  
212.     }  
213.   
214.   
215.     //获取当前阻塞队列大小  
216.     public int size() {  
217.         lock.lock();  
218.         try {  
219.             return queue.size();  
220.         } finally {  
221.             lock.unlock();  
222.         }  
223.     }  
224.   
225.     public void tryPut(RejectPolicy<T> rejectPolicy, T task) {  
226.         lock.lock();  
227.         try {  
228.             //判断队列是否已满  
229.             if (queue.size() == capcity) {  
230.                 rejectPolicy.reject(this, task);  
231.             } else {  
232.                 queue.addLast(task);  
233.                 emptyWaitSet.signal();  
234.             }  
235.         } finally {  
236.             lock.unlock();  
237.         }  
238.     }  
239. }

复制代码

invokeAll方法可以传入任务的集合，同样的任务的返回值也会以列表形式返回
ThreadPoolExecutor-invokeAny

1. public static void main(String[] args) throws ExecutionException, InterruptedException {  
2.     ExecutorService pool = Executors.newFixedThreadPool(2);  
3.     Future<String> result = pool.submit(() -> {  
4.         System.out.println("running");  
5.         Thread.sleep(1000);  
6.         return "ok";  
7.     });  
8.     System.out.println(result.get());  
9. }

复制代码

invokeAny方法同样可以传入任务的集合，只不过最后返回的结果并不是任务的结果集合，而是最早完成的那个任务的结果。
ThreadPoolExecutor-shutdown

1. public static void main(String[] args) throws ExecutionException, InterruptedException {  
2.     ExecutorService pool = Executors.newFixedThreadPool(2);  
3.     List<Future<String>> results = pool.invokeAll(Arrays.asList(() -> {  
4.                 System.out.println("begin");  
5.                 Thread.sleep(1000);  
6.                 return "1";  
7.             },  
8.             () -> {  
9.                 System.out.println("begin");  
10.                 Thread.sleep(500);  
11.                 return "2";  
12.             }));  
13.     results.forEach(f -> {  
14.         try {  
15.             System.out.printf(f.get());  
16.         } catch (InterruptedException e) {  
17.             e.printStackTrace();  
18.         } catch (ExecutionException e) {  
19.             e.printStackTrace();  
20.         }  
21.     });  
22. }

复制代码

shutdown方法会将线程池的状态变为SHUTDOWN

• 不会接受新任务
  
• 但已提交的任务会执行完
  
• 此方法不会阻塞调用线程的执行
  

ThreadPoolExecutor-shutdownNow

1. public static void main(String[] args) throws ExecutionException, InterruptedException {  
2.     ExecutorService pool = Executors.newFixedThreadPool(2);  
3.     String result = pool.invokeAny(Arrays.asList(() -> {  
4.                 System.out.println("begin");  
5.                 Thread.sleep(1000);  
6.                 return "1";  
7.             },  
8.             () -> {  
9.                 System.out.println("begin");  
10.                 Thread.sleep(500);  
11.                 return "2";  
12.             }));  
13.     pool.awaitTermination(1000, TimeUnit.MILLISECONDS);  
14.     System.out.println(result);  
15. }

复制代码

shutdownNow方法会将线程池状态变为STOP

• 不会接受新任务
  
• 会将队列中现有的任务返回
  
• 并且用interrupt方法中断正在执行的任务