配置ThreadPoolExecutor(java并发编程第8章)

朱雀 2022-07-12 08:00 149阅读 0赞

Executors的静态方法newCachedThreadPool, newFixedThreadPool, newScheduledThreadPool所返回的线程池都是ThreadPoolExecutor对象或者其子类对象. ThreadPoolExecutor提供了多种配置, 可以根据实际定制合适的线程池.

线程的创建和销毁

ThreadPoolExecutor构造函数中的corePoolSize, maximumPoolSize, keepAliveTime参数与线程的创建和销毁相关.

corePoolSize指定ThreadPoolExecutor中持有的核心线程数, 除非task队列已满, ThreadPoolExecutor不会创建超过核心线程数的线程(corePoolSize为0时是一种特殊情况, 此时就算task队列没有饱和, 向线程池第一次提交task时仍然会创建新的线程), 核心线程一旦创建就不会销毁, 除非设置了allowCoreThreadTimeOut(true), 或者关闭线程池.

maximumPoolSize指定线程池中持有的最大线程数. 对于超过核心线程数的线程, 如果在指定的超时时间内没有使用到, 就会被销毁.

keepAliveTime指定超时时间.

Executors类的静态方法创建线程池的源码:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **static** ExecutorService newCachedThreadPool() \{  
2.      // 核心线程数为0, 最大线程数为Integer.MAX\_VALUE, 超时时间为60s  
3.      **return** **new** ThreadPoolExecutor(0, Integer.MAX\_VALUE, 60L, TimeUnit.SECONDS, **new** SynchronousQueue<Runnable>());  
4.  \}  
5.    
6.  **public** **static** ExecutorService newFixedThreadPool(**int** nThreads) \{  
7.      // 核心线程数和最大线程数都为调用方指定的值nThreads, 超时时间为0  
8.      **return** **new** ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,  
9.              **new** LinkedBlockingQueue<Runnable>());  
10. \}  
11.   
12. **public** **static** ScheduledExecutorService newScheduledThreadPool(**int** corePoolSize) \{  
13.     // 核心线程数由调用方指定, 最大线程数为Integer.MAX\_VALUE, 超时时间为0  
14.     **return** **new** ThreadPoolExecutor(corePoolSize, Integer.MAX\_VALUE, 0, TimeUnit.NANOSECONDS, **new** DelayedWorkQueue());  
15. \}

task队列

线程池内部持有一个task队列, 当task的提交速度超过task的执行速度时, task将被缓存在task队列中等待有线程可用时再执行. ThreadPoolExecutor在创建时可以为其指定task队列, 开发者一般有三种选择: 有界队列, 无界队列以及同步队列. Executors.newFixedThreadPool和Executors.newScheduledThreadPool返回的ThreadPoolExecutor对象使用的是无界队列, 而Executors.newCashedThreadPool返回的ThreadPoolExecutor对象使用的是同步队列.

为线程数不多的线程池指定一个容量大的队列(或者无界队列), 有助于减少线程间切换, CPU等方面的消耗, 代价是可能会造成吞吐量下降. 如果使用的是有界队列, 队列可能会被填满, 此时将根据指定的饱和策略进行处理(见之后的讲述).

对于线程数很大的线程池, 可以使用同步队列. 同步队列(SynchronousQueue)其实不能算是一种队列, 因为同步队列没有缓存的作用. 使用同步队列时, task被提交时, 直接由线程池中的线程接手. 如果此时线程池中没有可用的线程, 线程池将创建新的线程接手. 如果线程池无法创建新的线程(比如线程数已到达maximumPoolSize), 则根据指定的饱和策略进行处理(同样见之后的讲述).

饱和策略

如果线程池使用的是有界队列, 那么当有界队列满时继续提交task时饱和策略会被触发.

如果线程池使用的是同步队列, 那么当线程池无法创建新的线程接手task时饱和策略会被触发.

如果线程池被关闭后, 仍然向其提交task时, 饱和策略也会被触发.

ThreadPoolExecutor.setRejectedExecutionHandler方法用于设定饱和策略. 该方法接受一个RejectedExecutionHandler对象作为参数. RejectedExecutionHandler只定义了一个方法:rejectedExecution(Runnable r, ThreadPoolExecutor executor). rejectedExecution方法在饱和策略被触发时由系统回调.

ThreadPoolExecutor类中预定义了多个RejectedExecutionHandler的实现类: AbortPolicy, CallerRunsPolicy, DiscardPolicy, 和DiscardOldestPolicy.

AbortPolicy是默认的饱和策略, 其rejectedExecution方法为:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **void** rejectedExecution(Runnable r, ThreadPoolExecutor e) \{  
2.      **throw** **new** RejectedExecutionException();  
3.  \}

可见默认情况下, 触发饱和策略时将抛出RejectedExecutionException异常.

CallerRunsPolicy. 饱和时将在提交task的线程中执行task, 而不是由线程池中的线程执行:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **void** rejectedExecution(Runnable r, ThreadPoolExecutor e) \{  
2.      **if** (!e.isShutdown()) \{  
3.          r.run();  
4.      \}  
5.  \}

使用CallerRunsPolicy的例子:

Java代码   ![收藏代码][icon_star.png]

1.  **class** LifecycleWebServer \{  
2.      // MAX\_THREAD\_COUNT和MAX\_QUEUE\_COUNT的值根据系统的实际情况确定  
3.      **private** **static** **final** **int** MAX\_THREAD\_COUNT = 100;  
4.      **private** **static** **final** **int** MAX\_QUEUE\_COUNT = 1000;  
5.    
6.      // 使用有界队列作为task队列, 当有界队列满时, 将触发饱和策略  
7.      **private** **final** ThreadPoolExecutor exec = **new** ThreadPoolExecutor(0, MAX\_THREAD\_COUNT, 60L, TimeUnit.SECONDS,  
8.              **new** ArrayBlockingQueue<Runnable>(MAX\_QUEUE\_COUNT));  
9.    
10.     **public** **void** start() **throws** IOException \{  
11.         // 设置饱和策略为CallerRunsPolicy  
12.         exec.setRejectedExecutionHandler(**new** ThreadPoolExecutor.CallerRunsPolicy());  
13.         ServerSocket socket = **new** ServerSocket(80);  
14.         **while** (!exec.isShutdown()) \{  
15.             **try** \{  
16.                 **final** Socket conn = socket.accept();  
17.                 exec.execute(**new** Runnable() \{  
18.                     **public** **void** run() \{  
19.                         handleRequest(conn);  
20.                     \}  
21.                 \});  
22.             \} **catch** (RejectedExecutionException e) \{  
23.                 **if** (!exec.isShutdown())  
24.                     log("task submission rejected", e);  
25.             \}  
26.         \}  
27.     \}  
28.   
29.     **public** **void** stop() \{  
30.         exec.shutdown();  
31.     \}  
32.   
33.     **void** handleRequest(Socket connection) \{  
34.         Request req = readRequest(connection);  
35.         **if** (isShutdownRequest(req))  
36.             stop();  
37.         **else**  
38.             dispatchRequest(req);  
39.     \}  
40.       
41.     **public** **static** **void** main(String\[\] args) \{  
42.         LifecycleWebServer server = **new** LifecycleWebServer();  
43.         **try** \{  
44.             // 在main线程中启动server  
45.             server.start();  
46.         \} **catch** (IOException e) \{  
47.             e.printStackTrace();  
48.         \}  
49.     \}  
50. \}

LifecycleWebServer中的线程池使用CallerRunsPolicy作为其饱和策略. 如果线程池饱和时main线程仍然向线程池提交task, 那么task将在main中执行. main线程执行task是需要一定时间的, 这样就给了线程池喘息的机会, 而且main线程在执行task的时间内无法接受socket连接, 因此socket连接请求将缓存在tcp层. 如果server过载持续的时间较长, 使得tcp层的缓存不够, 那么tcp缓存将根据其策略丢弃部分请求. 如此一来, 整个系统的过载压力逐步向外扩散: 线程池-线程池中的队列-main线程-tcp层-client. 这样的系统在发生过载时是比较优雅的: 既不会因为过多的请求而导致系统资源耗尽, 也不会一发生过载时就拒绝服务, 只有发生长时间系统过载时才会出现客户端无法连接的情况.

DiscardPolicy. 该策略将最新提交的task丢弃:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **void** rejectedExecution(Runnable r, ThreadPoolExecutor e) \{  
2.      // 丢弃, 不做任何处理  
3.  \}

DiscardOldestPolicy. 该策略丢弃队列中处于对头的task, 且试着再次提交最新的task:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **void** rejectedExecution(Runnable r, ThreadPoolExecutor e) \{  
2.      **if** (!e.isShutdown()) \{  
3.      e.getQueue().poll();  
4.      e.execute(r);  
5.      \}  
6.  \}

DiscardOldestPolicy与PriorityBlockingQueue结合使用时可能会造成不好的结果, 因为PriorityBlockingQueue中位于对头的task是优先级最高的task, 发生饱和时反而首先丢弃优先级高的task可能不符合需求.

ThreadPoolExecutor没有提供饱和时阻塞的策略, 不过开发者可以结合Semaphore实现:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** BoundedExecutor \{  
2.      **private** **final** Executor exec;  
3.      **private** **final** Semaphore semaphore;  
4.    
5.      **public** BoundedExecutor(Executor exec, **int** bound) \{  
6.          **this**.exec = exec;  
7.          // 设定信号量permit的上限  
8.          **this**.semaphore = **new** Semaphore(bound);  
9.      \}  
10.   
11.     **public** **void** submitTask(**final** Runnable command) **throws** InterruptedException \{  
12.         // 提交task前先申请permit, 如果无法申请到permit, 调用submitTask的线程将被阻塞, 直到有permit可用  
13.         semaphore.acquire();  
14.         **try** \{  
15.             exec.execute(**new** Runnable() \{  
16.                 **public** **void** run() \{  
17.                     **try** \{  
18.                         command.run();  
19.                     \} **finally** \{  
20.                         // 提交成功了, 运行task后释放permit  
21.                         semaphore.release();  
22.                     \}  
23.                 \}  
24.             \});  
25.         \} **catch** (RejectedExecutionException e) \{  
26.             // 如果没有提交成功, 也需要释放permit  
27.             semaphore.release();  
28.         \}  
29.     \}  
30. \}

ThreadFactory

在创建ThreadPoolExecutor时还可以为其指定ThreadFactory, 当线程池需要创建新的线程时会调用ThreadFactory的newThread方法. 默认的ThreadFactory创建的线程是nonDaemon, 线程优先级为NORM\_PRIORITY的线程, 并且为其指定了可识别的线程名称:

Java代码   ![收藏代码][icon_star.png]

1.  **public** Thread newThread(Runnable r) \{  
2.      Thread t = **new** Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);  
3.      **if** (t.isDaemon())  
4.          t.setDaemon(**false**);  
5.      **if** (t.getPriority() != Thread.NORM\_PRIORITY)  
6.          t.setPriority(Thread.NORM\_PRIORITY);  
7.      **return** t;  
8.  \}

开发者可以根据自身需要为ThreadPoolExecutor指定自定义的ThreadFactory. 例如:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** MyThreadFactory **implements** ThreadFactory \{  
2.      **private** **final** String poolName;  
3.    
4.      **public** MyThreadFactory(String poolName) \{  
5.          **this**.poolName = poolName;  
6.      \}  
7.    
8.      **public** Thread newThread(Runnable runnable) \{  
9.          **return** **new** MyAppThread(runnable, poolName);  
10.     \}  
11. \}  
12.   
13. **public** **class** MyAppThread **extends** Thread \{  
14.     **public** **static** **final** String DEFAULT\_NAME = "MyAppThread";  
15.     **private** **static** **volatile** **boolean** debugLifecycle = **false**;  
16.     **private** **static** **final** AtomicInteger created = **new** AtomicInteger();  
17.     **private** **static** **final** AtomicInteger alive = **new** AtomicInteger();  
18.     **private** **static** **final** Logger log = Logger.getAnonymousLogger();  
19.   
20.     **public** MyAppThread(Runnable r) \{  
21.         **this**(r, DEFAULT\_NAME);  
22.     \}  
23.   
24.     **public** MyAppThread(Runnable runnable, String name) \{  
25.         // 为自定义的Thread类指定线程名称  
26.         **super**(runnable, name + "-" + created.incrementAndGet());  
27.         // 设置UncaughtExceptionHandler. UncaughtExceptionHandler的uncaughtException方法将在线程运行中抛出未捕获异常时由系统调用  
28.         setUncaughtExceptionHandler(**new** Thread.UncaughtExceptionHandler() \{  
29.             **public** **void** uncaughtException(Thread t, Throwable e) \{  
30.                 log.log(Level.SEVERE, "UNCAUGHT in thread " + t.getName(), e);  
31.             \}  
32.         \});  
33.     \}  
34.   
35.     **public** **void** run() \{  
36.         // Copy debug flag to ensure consistent value throughout.   
37.         **boolean** debug = debugLifecycle;  
38.         **if** (debug)  
39.             log.log(Level.FINE, "Created " + getName());  
40.         **try** \{  
41.             alive.incrementAndGet();  
42.             **super**.run();  
43.         \} **finally** \{  
44.             alive.decrementAndGet();  
45.             **if** (debug)  
46.                 log.log(Level.FINE, "Exiting " + getName());  
47.         \}  
48.     \}  
49.   
50.     **public** **static** **int** getThreadsCreated() \{  
51.         **return** created.get();  
52.     \}  
53.   
54.     **public** **static** **int** getThreadsAlive() \{  
55.         **return** alive.get();  
56.     \}  
57.   
58.     **public** **static** **boolean** getDebug() \{  
59.         **return** debugLifecycle;  
60.     \}  
61.   
62.     **public** **static** **void** setDebug(**boolean** b) \{  
63.         debugLifecycle = b;  
64.     \}  
65. \}

扩展ThreadPoolExecutor

ThreadPoolExecutor类提供了多个"钩子"方法, 以供其子类实现, 比如beforeExecute, afterExecute, terminated等. 所谓"钩子"是指基类预留的, 但是没有提供具体实现的方法, 其方法体为空. 子类可以根据需要为"钩子"提供具体实现.

beforeExecute和afterExecute方法分别在执行task前后调用:

Java代码   ![收藏代码][icon_star.png]

1.  **private** **void** runTask(Runnable task) \{  
2.      **final** ReentrantLock runLock = **this**.runLock;  
3.      runLock.lock();  
4.      **try** \{  
5.          **if** (runState < STOP && Thread.interrupted() && runState >= STOP)  
6.              thread.interrupt();  
7.          **boolean** ran = **false**;  
8.          beforeExecute(thread, task);  
9.          **try** \{  
10.             task.run();  
11.             ran = **true**;  
12.             afterExecute(task, **null**);  
13.             ++completedTasks;  
14.         \} **catch** (RuntimeException ex) \{  
15.             **if** (!ran)  
16.                 afterExecute(task, ex);  
17.             **throw** ex;  
18.         \}  
19.     \} **finally** \{  
20.         runLock.unlock();  
21.     \}  
22. \}

beforeExecute和afterExecute方法可以用于记录日志, 统计数据等操作.

terminated方法在线程池被关闭后调用. terminated方法可以用于释放线程池申请的资源.

[icon_star.png]: /images/20220711/8a7ef72daf034269b5014d4f1f6b030f.png