关于线程池ThreadPoolExecutor使用总结

 1 import java.io.Serializable;
 2 import java.util.concurrent.ArrayBlockingQueue;
 3 import java.util.concurrent.ThreadPoolExecutor;
 4 import java.util.concurrent.TimeUnit;
 5 
 6 public class TestThreadPool {
 7 
 8     private static int produceTaskSleepTime = 2;
 9 
10     private static int produceTaskMaxNumber = 10;
11 
12     public static void main(String[] args) {
13         // 构造一个线程池
14         ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3),
15                 new ThreadPoolExecutor.DiscardOldestPolicy());
16 //        corePoolSize： 线程池维护线程的最少数量 ==============>2
17 //        maximumPoolSize：线程池维护线程的最大数量 ==============>4 
18 //        keepAliveTime： 线程池维护线程所允许的空闲时间 ==============>3
19 //        unit： 线程池维护线程所允许的空闲时间的单位 ==============>TimeUnit.SECONDS
20 //        workQueue： 线程池所使用的缓冲队列 ==============>new ArrayBlockingQueue<Runnable>(3)
21 //        handler： 线程池对拒绝任务的处理策略 ==============>new ThreadPoolExecutor.DiscardOldestPolicy()
22 
23         for (int i = 1; i <= produceTaskMaxNumber; i++) {
24             try {
25                 // 产生一个任务，并将其加入到线程池
26                 String task = "task@ " + i;
27                 System.out.println("put " + task);
28                 threadPool.execute(new ThreadPoolTask(task));
29 
30                 // 便于观察，等待一段时间
31                 Thread.sleep(produceTaskSleepTime);
32             } catch (Exception e) {
33                 e.printStackTrace();
34             }
35         }
36     }
37 }
38 
39 /**
40  * 线程池执行的任务
41  */
42 class ThreadPoolTask implements Runnable, Serializable {
43 
44     private static final long serialVersionUID = 0;
45 
46     private static int consumeTaskSleepTime = 2000;
47 
48     // 保存任务所需要的数据
49     private Object threadPoolTaskData;
50 
51     ThreadPoolTask(Object tasks) {
52         this.threadPoolTaskData = tasks;
53     }
54 
55     public void run() {
56         // 处理一个任务，这里的处理方式太简单了，仅仅是一个打印语句
57         System.out.println(Thread.currentThread().getName());
58         System.out.println("start .." + threadPoolTaskData);
59 
60         try {
61             //便于观察，等待一段时间
62             Thread.sleep(consumeTaskSleepTime);
63         } catch (Exception e) {
64             e.printStackTrace();
65         }
66         threadPoolTaskData = null;
67     }
68 
69     public Object getTask() {
70         return this.threadPoolTaskData;
71     }
72 }

 1 import java.util.Queue;
 2 import java.util.concurrent.ArrayBlockingQueue;
 3 import java.util.concurrent.ThreadPoolExecutor;
 4 import java.util.concurrent.TimeUnit;
 5 
 6 public class ThreadPoolExecutorTest {
 7 
 8     private static int queueDeep = 4;
 9 
10     public void createThreadPool() {
11         /*
12          * 创建线程池，最小线程数为2，最大线程数为4，线程池维护线程的空闲时间为3秒， 使用队列深度为4的有界队列，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，
13          * 然后重试执行程序（如果再次失败，则重复此过程），里面已经根据队列深度对任务加载进行了控制。
14          */
15         ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueDeep),
16                 new ThreadPoolExecutor.DiscardOldestPolicy());
17 
18         // 向线程池中添加 10 个任务
19         for (int i = 0; i < 10; i++) {
20             try {
21                 Thread.sleep(1);
22             } catch (InterruptedException e) {
23                 e.printStackTrace();
24             }
25             while (getQueueSize(tpe.getQueue()) >= queueDeep) {
26                 System.out.println("队列已满，等3秒再添加任务");
27                 try {
28                     Thread.sleep(3000);
29                 } catch (InterruptedException e) {
30                     e.printStackTrace();
31                 }
32             }
33             TaskThreadPool ttp = new TaskThreadPool(i);
34             System.out.println("put i:" + i);
35             tpe.execute(ttp);
36         }
37 
38         tpe.shutdown();
39     }
40 
41     private synchronized int getQueueSize(Queue queue) {
42         return queue.size();
43     }
44 
45     public static void main(String[] args) {
46         ThreadPoolExecutorTest test = new ThreadPoolExecutorTest();
47         test.createThreadPool();
48     }
49 
50     class TaskThreadPool implements Runnable {
51 
52         private int index;
53 
54         public TaskThreadPool(int index) {
55             this.index = index;
56         }
57 
58         public void run() {
59             System.out.println(Thread.currentThread() + " index:" + index);
60             try {
61                 Thread.sleep(3000);
62             } catch (InterruptedException e) {
63                 e.printStackTrace();
64             }
65         }
66     }
67 }