线程池系列三：ThreadPoolExecutor讲解

Posted 2020-08-28 672530440

 

三、一个用队列处理线程池例子

 

Java代码  

1. package demo;  
2.   
3. import java.util.Queue;  
4. import java.util.concurrent.ArrayBlockingQueue;  
5. import java.util.concurrent.ThreadPoolExecutor;  
6. import java.util.concurrent.TimeUnit;  
7.   
8. public class ThreadPoolExecutorTest  
9. {  
10.   
11.     private static int queueDeep = 4;  
12.   
13.     public void createThreadPool()  
14.     {  
15.         /*   
16.          * 创建线程池，最小线程数为2，最大线程数为4，线程池维护线程的空闲时间为3秒，   
17.          缓冲队列为4，线程执行时间是3秒。   
18.          */   
19.         ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueDeep),  
20.                 new ThreadPoolExecutor.DiscardOldestPolicy()); //这里采取的是抛弃旧的任务  
21.   
22.   
23.         // 向线程池中添加 10 个任务  
24.         for (int i = 0; i < 10; i++)  
25.         {  
26.             try  
27.             {  
28.                 Thread.sleep(1);  
29.             }  
30.             catch (InterruptedException e)  
31.             {  
32.                 e.printStackTrace();  
33.             }  
34.             while (getQueueSize(tpe.getQueue()) >= queueDeep)  
35.             {  
36.                 System.out.println("队列已满，等3秒再添加任务");  
37.                 try  
38.                 {  
39.                     Thread.sleep(3000);  
40.                 }  
41.                 catch (InterruptedException e)  
42.                 {  
43.                     e.printStackTrace();  
44.                 }  
45.             }  
46.             TaskThreadPool ttp = new TaskThreadPool(i);  
47.             System.out.println("put i:" + i);  
48.             tpe.execute(ttp);  
49.         }  
50.   
51.         tpe.shutdown();  
52.     }  
53.   
54.     private synchronized int getQueueSize(Queue queue)  
55.     {  
56.         return queue.size();  
57.     }  
58.   
59.     public static void main(String[] args)  
60.     {  
61.         ThreadPoolExecutorTest test = new ThreadPoolExecutorTest();  
62.         test.createThreadPool();  
63.     }  
64.   
65.     class TaskThreadPool implements Runnable  
66.     {  
67.         private int index;  
68.   
69.         public TaskThreadPool(int index)  
70.         {  
71.             this.index = index;  
72.         }  
73.   
74.         public void run()  
75.         {  
76.             System.out.println(Thread.currentThread() + " index:" + index);  
77.             try  
78.             {  
79.                 Thread.sleep(3000);  
80.             }  
81.             catch (InterruptedException e)  
82.             {  
83.                 e.printStackTrace();  
84.             }  
85.         }  
86.     }  
87. }  

 

说明：

 

这里执行的结果为：

put i:0　　　　　　　　　　　　　　　　　　　　　　0加入进去池子
Thread[pool-1-thread-1,5,main] index:0　　　　　0开始执行
put i:1　　　　　　　　　　　　　　　　　　　　　　1加入进去池子
Thread[pool-1-thread-2,5,main] index:1    　　　  1开始执行
put i:2　　　　　　　　　　　　　　　　　　　　　　2加入缓冲
put i:3　　　　　　　　　　　　　　　　　　　　　　3加入缓冲
put i:4　　　　　　　　　　　　　　　　　　　　　　4加入缓冲
put i:5　　　　　　　　　　　　　　　　　　　　　　5加入缓冲
队列已满，等3秒再添加任务

（上述代码首先一次性执行，会在这里停一会，因为添加线程不需要时间）
Thread[pool-1-thread-1,5,main] index:2　　　　2开始执行
Thread[pool-1-thread-2,5,main] index:3　　　　3开始执行
put i:6　　　　　　　　　　　　　　　　　　　　　6加入缓冲
put i:7　　　　　　　　　　　　　　　　　　　　　7加入缓冲
队列已满，等3秒再添加任务

（会停一会，因为任务执行要3秒时间，2,3,4,5会先执行2,3 ，因为2,3先进入缓冲队列）
Thread[pool-1-thread-1,5,main] index:4
Thread[pool-1-thread-2,5,main] index:5
put i:8
put i:9

（停一会，任务执行要3秒，）
Thread[pool-1-thread-1,5,main] index:6
Thread[pool-1-thread-2,5,main] index:7

（要停一会，任务执行要3秒）
Thread[pool-1-thread-1,5,main] index:8
Thread[pool-1-thread-2,5,main] index:9

 （执行完毕）

ps:这里是当队列已满时线程就一直等待了，不会再新创建线程，所以一直就只有1和2两个线程来执行。

 

如果把

Java代码  

1. while (getQueueSize(tpe.getQueue()) >= queueDeep){}  

 这一段去掉，那么执行结果为：

put i:0
Thread[pool-1-thread-1,5,main] index:0
put i:1
Thread[pool-1-thread-2,5,main] index:1
put i:2
put i:3
put i:4
put i:5
put i:6
Thread[pool-1-thread-3,5,main] index:6
put i:7
Thread[pool-1-thread-4,5,main] index:7
put i:8
put i:9
Thread[pool-1-thread-1,5,main] index:4
Thread[pool-1-thread-2,5,main] index:5
Thread[pool-1-thread-3,5,main] index:8
Thread[pool-1-thread-4,5,main] index:9

ps:这个执行顺序是0，1两个任务先进来，分别由线程1,2来执行，然后2,-5进来，队列满，6任务进来，因为队列已满，且1,2线程还未执行完，没有可用的线程，所以创建新的线程来运行6。7任务同理。然后8任务进来，队列已满，且1,2,3,4线程未执行完，线程数又等于了最多4个线程的限制，这时看线程池的执行策略为DiscardOldestPolicy,就是抛弃旧的任务，故开始进队列的2任务被抛弃，3任务同理，8,9任务进入队列，然后这时1-4线程已经执行完自己的任务，开始执行队列中的4,5,8,9

如果更改执行策略，那么相应的结果也会不一样，如果不希望有任务被抛弃，那么可以采用CallerRunsPolicy()策略。