線程池的管理哲學(xué)
收到不少讀者反饋，其中有這幾道關(guān)于線程池的問題：
- 在進(jìn)行線程池設(shè)計(jì)時(shí)，如何選擇拒絕策略？
- 如果不允許丟棄任務(wù)任務(wù)，應(yīng)該選擇哪個(gè)拒絕策略？
- 使用CallerRunsPolicy這個(gè)拒絕策略有什么風(fēng)險(xiǎn)？有沒有更好的處理方式呢？
一、詳解拒絕策略常見問題
1. 線程池是如何工作的
我們先來復(fù)習(xí)一下線程池的工作流程，每次任務(wù)提交時(shí)，線程池都會(huì)嘗試將任務(wù)提交到核心線程上，如果線程數(shù)小于核心線程數(shù)，線程池就會(huì)添加工作線程并執(zhí)行當(dāng)前任務(wù)。 若核心線程都處于工作狀態(tài)，這就表明當(dāng)前線程池有些忙碌，那么這些無法及時(shí)處理的任務(wù)就會(huì)提交到阻塞任務(wù)隊(duì)列中。 隨著任務(wù)的遞增，任務(wù)隊(duì)列無法容納最新的任務(wù)，線程池就會(huì)認(rèn)為現(xiàn)處于高峰期，便臨時(shí)增加應(yīng)急線程處理任務(wù)。隨著任務(wù)逐步處理完成，線程在指定時(shí)間內(nèi)沒有要處理的任務(wù)，這些線程也就會(huì)依次退出。
對(duì)應(yīng)我們也給出ThreadPoolExecutor提交任務(wù)的execute方法的源碼：
public void execute(Runnable command) {
//任務(wù)判空
if (command == null)
throw new NullPointerException();
//查看當(dāng)前運(yùn)行的線程數(shù)量
int c = ctl.get();
//若小于核心線程則直接添加一個(gè)工作線程并執(zhí)行任務(wù)
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//如果線程數(shù)等于核心線程數(shù)則嘗試將任務(wù)入隊(duì)
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//入隊(duì)失敗，調(diào)用addWorker參數(shù)為false，嘗試創(chuàng)建應(yīng)急線程處理突發(fā)任務(wù)
else if (!addWorker(command, false))
//如果創(chuàng)建應(yīng)急線程失敗，說明當(dāng)前線程數(shù)已經(jīng)大于最大線程數(shù)，這個(gè)任務(wù)只能拒絕了
reject(command);
}上文提到了應(yīng)急線程長(zhǎng)時(shí)間沒有要處理的任務(wù)就會(huì)被銷毀的邏輯，這里我們也簡(jiǎn)單的介紹一下，首先在線程池中每一個(gè)線程是以Worker的形式封裝呈現(xiàn)，其本質(zhì)就是對(duì)Thread的封裝，Worker啟動(dòng)后會(huì)調(diào)用run方法調(diào)用runWorker方法輪詢處理任務(wù)：
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
final Thread thread;
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
//......
}查看runWorker方法，一旦在規(guī)定時(shí)間內(nèi)getTask沒有拿到任務(wù)就會(huì)退出循環(huán)，直接通過processWorkerExit結(jié)束這個(gè)工作線程：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//對(duì)應(yīng)時(shí)間內(nèi)沒有拿到task則退出循環(huán)
while (task != null || (task = getTask()) != null) {
//略
}
completedAbruptly = false;
} finally {
//結(jié)束這個(gè)工作線程
processWorkerExit(w, completedAbruptly);
}
}自此，我們將線程池整體工作流程簡(jiǎn)單的梳理完畢。
2. 拒絕策略的選擇
先來說說第一道題，關(guān)于拒絕策略的選擇，我們不妨直接查看RejectedExecutionHandler子類的源碼進(jìn)行說明。 先來看看CallerRunsPolicy，該拒絕策略會(huì)直接用當(dāng)前調(diào)用者執(zhí)行當(dāng)前任務(wù)：
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
//直接基于當(dāng)前線程調(diào)用run方法執(zhí)行任務(wù)
r.run();
}
}
}然后就是AbortPolicy，也很簡(jiǎn)單，直接拋異常：
public static class AbortPolicy implements RejectedExecutionHandler {
/**
* Creates an {@code AbortPolicy}.
*/
public AbortPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}DiscardPolicy則是什么也不做，這也就意味著這個(gè)任務(wù)沒有任務(wù)處理，等同于丟棄：
public static class DiscardPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardPolicy}.
*/
public DiscardPolicy() { }
//不做任何事情任務(wù)直接丟棄
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}最后一個(gè)就是DiscardOldestPolicy，該策略會(huì)將隊(duì)首部任務(wù)丟棄，然后嘗試將再次execute這個(gè)任務(wù)：
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
//丟掉隊(duì)首的任務(wù)，然后往線程池提交當(dāng)前任務(wù)
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}不同拒絕策略都有著不同的使用場(chǎng)景：
- 如果我們的任務(wù)不算耗時(shí)還要保證能夠被執(zhí)行，那么CallerRunsPolicy則是第一選擇。
- 若突增大量任務(wù)導(dǎo)致無法及時(shí)處理從業(yè)務(wù)的角度認(rèn)為是異常的話，那么我們則建議拋出AbortPolicy讓開發(fā)介入及時(shí)調(diào)優(yōu)處理，前提是當(dāng)前業(yè)務(wù)正處于業(yè)務(wù)提測(cè)階段。
- 對(duì)于那些需要提交實(shí)時(shí)性消息的監(jiān)控型任務(wù)，那么新提交的任務(wù)勢(shì)必實(shí)時(shí)性會(huì)由于更早的任務(wù)，這種場(chǎng)景使用DiscardOldestPolicy即可。
- 如果這些任務(wù)相較于系統(tǒng)可靠性來說，如果不是很重要，那么直接采用rejectedExecution丟棄任務(wù)即可。
3. 主流框架對(duì)于拒絕策略的選擇
只要繼承RejectedExecutionHandler就可以實(shí)現(xiàn)相應(yīng)的拒絕策略，所以我們也不妨看看一些主流的框架是如何使用拒絕策略的吧。
tomcat線程池的拒絕策略也是拋出異常：
private static class RejectHandler implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r,
java.util.concurrent.ThreadPoolExecutor executor) {
throw new RejectedExecutionException();
}
}而Dubbo則相對(duì)友好一些，它會(huì)優(yōu)先打印一個(gè)日志，并告知異常堆棧信息，然后拋出異常：
public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
//......
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
String msg = String.format("Thread pool is EXHAUSTED!" +
" Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d)," +
" Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
url.getProtocol(), url.getIp(), url.getPort());
logger.warn(msg);
dumpJStack();
throw new RejectedExecutionException(msg);
}
private void dumpJStack() {
//省略實(shí)現(xiàn)
}
}Netty就相對(duì)穩(wěn)健一些，它的拒絕策略則是直接創(chuàng)建一個(gè)線程池以外的線程處理這些任務(wù)，為了保證任務(wù)的實(shí)時(shí)處理，這種做法可能需要良好的硬件設(shè)備且臨時(shí)創(chuàng)建的線程無法做到準(zhǔn)確的監(jiān)控：
private static final class NewThreadRunsPolicy implements RejectedExecutionHandler {
NewThreadRunsPolicy() {
super();
}
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
try {
final Thread t = new Thread(r, "Temporary task executor");
t.start();
} catch (Throwable e) {
throw new RejectedExecutionException(
"Failed to start a new thread", e);
}
}
}ActiveMq則是嘗試在指定的時(shí)效內(nèi)盡可能的爭(zhēng)取將任務(wù)入隊(duì)，以保證最大交付：
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) {
try {
executor.getQueue().offer(r, 60, TimeUnit.SECONDS);
} catch (InterruptedException e) {
throw new RejectedExecutionException("Interrupted waiting for BrokerService.worker");
}
throw new RejectedExecutionException("Timed Out while attempting to enqueue Task.");
}
});4. CallerRunsPolicy存在的問題及解決對(duì)策
默認(rèn)情況下，我們都會(huì)為了保證任務(wù)不被丟棄都優(yōu)先考慮CallerRunsPolicy，這也是相對(duì)維穩(wěn)的做法，這種做法的隱患是假設(shè)走到CallerRunsPolicy的任務(wù)是個(gè)非常耗時(shí)的任務(wù)，就會(huì)導(dǎo)致主線程就很卡死。
下面就是筆者通過主線程使用線程池的方法，該線程池限定了最大線程數(shù)為2還有阻塞隊(duì)列大小為1，這意味著第4個(gè)任務(wù)就會(huì)走到拒絕策略：
//創(chuàng)建線程池
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1,
2,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1),
new ThreadPoolExecutor.CallerRunsPolicy());
threadPoolExecutor.execute(() -> {
log.info("核心線程執(zhí)行");
ThreadUtil.sleep(1, TimeUnit.DAYS);
});
threadPoolExecutor.execute(() -> {
log.info("任務(wù)入隊(duì)");
ThreadUtil.sleep(1, TimeUnit.DAYS);
});
threadPoolExecutor.execute(() -> {
log.info("應(yīng)急線程處理");
ThreadUtil.sleep(1, TimeUnit.DAYS);
});
threadPoolExecutor.execute(() -> {
log.info("CallerRunsPolicy task");
ThreadUtil.sleep(1, TimeUnit.DAYS);
});
threadPoolExecutor.execute(() -> {
log.info("因?yàn)橹骶€程卡住，無法被處理的任務(wù)");
});從輸出結(jié)果可以看出，因?yàn)镃allerRunsPolicy這個(gè)拒絕策略，導(dǎo)致耗時(shí)的任務(wù)用了主線程執(zhí)行，導(dǎo)致線程池阻塞，進(jìn)而導(dǎo)致后續(xù)任務(wù)無法及時(shí)執(zhí)行，嚴(yán)重的情況下很可能導(dǎo)致OOM：
2024-04-03 00:08:12.617 INFO 20804 --- [ main] com.sharkChili.ThreadPoolApplication : 啟動(dòng)成功?。?2024-04-03 00:08:15.739 INFO 20804 --- [pool-1-thread-1] com.sharkChili.ThreadPoolApplication : 核心線程執(zhí)行
2024-04-03 00:08:36.768 INFO 20804 --- [pool-1-thread-2] com.sharkChili.ThreadPoolApplication : 應(yīng)急線程處理
2024-04-03 00:08:49.333 INFO 20804 --- [ main] com.sharkChili.ThreadPoolApplication : CallerRunsPolicy task我們從問題的本質(zhì)入手，調(diào)用者采用CallerRunsPolicy是希望所有的任務(wù)都能夠被執(zhí)行，按照筆者的經(jīng)驗(yàn)，假如我們的場(chǎng)景是偶發(fā)這種突發(fā)場(chǎng)景，在內(nèi)存允許的情況下，我們建議增加阻塞隊(duì)列BlockingQueue的大小并調(diào)整堆內(nèi)存以容納更多的任務(wù)，確保任務(wù)能夠被準(zhǔn)確執(zhí)行。
若當(dāng)前服務(wù)器內(nèi)存資源緊張，但我們配置線程池還為盡可能利用到CPU，我們建議調(diào)整線程中maximumPoolSize以保證盡可能壓榨CPU資源：
如果服務(wù)器資源達(dá)到可利用的極限，這就意味我們要在設(shè)計(jì)策略上改變線程池的調(diào)度了，我們都知道，導(dǎo)致主線程卡死的本質(zhì)就是因?yàn)槲覀儾幌Ｍ魏我粋€(gè)任務(wù)被丟棄。換個(gè)思路，有沒有辦法既能保證任務(wù)不被丟棄且在服務(wù)器有余力時(shí)及時(shí)處理呢？
這里筆者提供的一種思路，即任務(wù)持久化，注意這里筆者更多強(qiáng)調(diào)的是思路而不是實(shí)現(xiàn)，這里所謂的任務(wù)持久化，包括但不限于:
- 設(shè)計(jì)一張任務(wù)表間任務(wù)存儲(chǔ)到MySQL數(shù)據(jù)庫(kù)中。
- Redis緩存任務(wù)。
- 將任務(wù)提交到消息隊(duì)列中。
筆者以方案一為例，通過繼承BlockingQueue實(shí)現(xiàn)一個(gè)混合式阻塞隊(duì)列，該隊(duì)列包含JDK自帶的ArrayBlockingQueue和一個(gè)自定義的隊(duì)列(數(shù)據(jù)表)，通過魔改隊(duì)列的添加邏輯達(dá)到任務(wù)可以存入ArrayBlockingQueue或者數(shù)據(jù)表的目的。
如此一來，一旦我們的線程池中線程達(dá)到滿載時(shí)，我們就可以通過拒絕策略將最新任務(wù)持久化到MySQL數(shù)據(jù)庫(kù)中，等到線程池有了有余力處理所有任務(wù)時(shí)，讓其優(yōu)先處理數(shù)據(jù)庫(kù)中的任務(wù)以避免"饑餓"問題。
這里筆者也給出混合隊(duì)列實(shí)現(xiàn)的核心源碼，即通過繼承BlockingQueue魔改了入隊(duì)和出隊(duì)的邏輯：
public class HybridBlockingQueue<E> implements BlockingQueue<E> {
private Object mysqlLock = new Object();
private ArrayBlockingQueue<E> arrayBlockingQueue;
//構(gòu)造方法初始化阻塞隊(duì)列大小
public HybridBlockingQueue(int maxSize) {
arrayBlockingQueue = new ArrayBlockingQueue<>(maxSize);
}
/**
* 線程池會(huì)調(diào)用的入隊(duì)方法
* @param e
* @return
*/
@Override
public boolean offer(E e) {
return arrayBlockingQueue.offer(e);
}
/**
* 取任務(wù)時(shí)，優(yōu)先從數(shù)據(jù)庫(kù)中讀取最早的任務(wù)
*
* @return
* @throws InterruptedException
*/
@Override
public E take() throws InterruptedException {
synchronized (mysqlLock) {
//從數(shù)據(jù)庫(kù)中讀取任務(wù)，通過上鎖讀取避免重復(fù)消費(fèi)
TaskInfoMapper taskMapper = SpringUtil.getBean(TaskInfoMapper.class);
TaskInfo taskInfo = taskMapper.selectByExample(null).stream()
.findFirst()
.orElse(null);
//若數(shù)據(jù)庫(kù)存在該任務(wù)，則先刪后返回
if (ObjUtil.isNotEmpty(taskInfo)) {
taskMapper.deleteByPrimaryKey(taskInfo.getId());
Task task = new Task(taskInfo.getData());
return (E) task;
}
}
//若數(shù)據(jù)庫(kù)沒有要處理的任務(wù)則從內(nèi)存中獲取
return arrayBlockingQueue.poll();
}
/**
* 帶有時(shí)間限制的任務(wù)獲取
*
* @param timeout
* @param unit
* @return
* @throws InterruptedException
*/
@Override
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
//從數(shù)據(jù)庫(kù)中讀取任務(wù)，通過上鎖讀取避免重復(fù)消費(fèi)
synchronized (mysqlLock) {
//從數(shù)據(jù)庫(kù)中讀取任務(wù)，
TaskInfoMapper taskMapper = SpringUtil.getBean(TaskInfoMapper.class);
TaskInfo taskInfo = taskMapper.selectByExample(null).stream()
.findFirst()
.orElse(null);
//若數(shù)據(jù)庫(kù)存在該任務(wù)，則先刪后返回
if (ObjUtil.isNotEmpty(taskInfo)) {
taskMapper.deleteByPrimaryKey(taskInfo.getId());
Task task = new Task(taskInfo.getData());
return (E) task;
}
}
//若數(shù)據(jù)庫(kù)沒有要處理的任務(wù)則從內(nèi)存中獲取
return arrayBlockingQueue.poll(timeout, unit);
}
//......
}接下來就是自定義拒絕策略了，很明顯我們的拒絕策略就叫持久化策略：
public class PersistentTaskPolicy implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
//任務(wù)入庫(kù)
TaskInfoMapper taskMapper = SpringUtil.getBean(TaskInfoMapper.class);
Task task = (Task) r;
TaskInfo taskInfo = new TaskInfo();
taskInfo.setData(JSONUtil.toJsonStr(task.getTaskInfo()));
taskMapper.insertSelective(taskInfo);
}
}最終我們的使用示例如下：
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1,
2,
60, TimeUnit.SECONDS,
new HybridBlockingQueue<>(1),
new PersistentTaskPolicy());
threadPoolExecutor.execute(new Task("core thread"));
threadPoolExecutor.execute(new Task("queueTask"));
threadPoolExecutor.execute(new Task("max thread"));
threadPoolExecutor.execute(new Task("insert into mysql database"));因?yàn)榫€程池?zé)o法及時(shí)處理而走了我們自定義的拒絕策略而持久化入庫(kù)，最終我們的insert into mysql database，等待線程池中其他任務(wù)完成后被取出執(zhí)行：
2024-04-14 11:30:16.865 INFO 1052 --- [ main] com.sharkChili.PersistentTaskPolicy : 任務(wù)持久化，taskInfo:{"data":"insert into mysql database"}
2024-04-14 11:31:08.516 INFO 1052 --- [pool-1-thread-2] com.sharkChili.Task : task execution completed,task info:max thread
2024-04-14 11:31:08.516 INFO 1052 --- [pool-1-thread-1] com.sharkChili.Task : task execution completed,task info:core thread
2024-04-14 11:32:08.563 INFO 1052 --- [pool-1-thread-1] com.sharkChili.Task : task execution completed,task info:queueTask
2024-04-14 11:32:08.563 INFO 1052 --- [pool-1-thread-2] com.sharkChili.Task : task execution completed,task info:insert into mysql database二、更高維度的思考——線程池限流的藝術(shù)
上文我們以大篇幅的維度探討拒絕策略上優(yōu)化，需要保證準(zhǔn)確、有效執(zhí)行的任務(wù)能夠被線程池處理，且不會(huì)破壞程序的穩(wěn)定性，即提交的任務(wù)能夠被正確處理且線程池不會(huì)被打死。
這一點(diǎn)，結(jié)合《Java并發(fā)編程實(shí)戰(zhàn)》的說法，我們也可以利用信號(hào)量Semaphore作為令牌，確保只有拿到令牌的線程才能將任務(wù)提交到線程池，保證線程池可以在單位時(shí)間內(nèi)按照我們?cè)O(shè)定的并發(fā)數(shù)執(zhí)行任務(wù)：
利用信號(hào)量完成線程池的限流，既保證任務(wù)可被執(zhí)行和工作線程池的穩(wěn)定性，又能將性能瓶頸和程序穩(wěn)定性問題拋給更高層級(jí)的調(diào)用者，尤上層根據(jù)需要決定當(dāng)前任務(wù)是等待被線程池處理，還是直接中斷結(jié)束。
對(duì)應(yīng)的我們給出流控性質(zhì)的線程池代碼示例，讀者可參考筆者所說的思路和注釋了解一下落地思路：
public class RateLimitedExecutor {
private final ExecutorService threadPool;
private final Semaphore semaphore;
//基于bound創(chuàng)建對(duì)應(yīng)并發(fā)度的線程池和流控令牌
public RateLimitedExecutor(int bound) {
this.threadPool = Executors.newFixedThreadPool(bound);
this.semaphore = new Semaphore(bound, true);
}
public void submitTask(final Runnable command) throws InterruptedException {
semaphore.acquire();
Console.log("{}獲取令牌成功，執(zhí)行時(shí)間:{}", Thread.currentThread().getName(), DateUtil.now());
try {
threadPool.execute(() -> {
try {
//執(zhí)行任務(wù)
command.run();
} finally {
//線程執(zhí)行完成后釋放令牌
semaphore.release();
}
});
} catch (RejectedExecutionException e) {//異常兜底
semaphore.release();
}
}
}對(duì)應(yīng)的我們也給出使用示例，可以看到我們創(chuàng)建了流控為5的線程池，并創(chuàng)建10個(gè)并發(fā)線程執(zhí)行提交操作：
public static void main(String[] args) {
RateLimitedExecutor executor = new RateLimitedExecutor(5);
for (int i = 0; i < 10; i++) {
new Thread(new Task("任務(wù)" + i, executor)).start();
}
}
private static class Task implements Runnable {
private final String threadName;
private final RateLimitedExecutor executor;
public Task(String threadName, RateLimitedExecutor executor) {
this.threadName = threadName;
this.executor = executor;
}
@SneakyThrows
@Override
public void run() {
executor.submitTask(() -> {
ThreadUtil.sleep(5000);
Console.log("{}執(zhí)行任務(wù)完成", threadName);
});
}
}輸出結(jié)果如下，可以看到流控符合預(yù)期為5，同時(shí)我們也將程序穩(wěn)定性和性能瓶頸等各方面的壓力轉(zhuǎn)移給上層調(diào)用者，避免了非必要的拒絕策略處理，讓線程池專注于并發(fā)度的優(yōu)化：
Thread-1獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:04
Thread-9獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:04
Thread-5獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:04
Thread-8獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:04
Thread-3獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:04
任務(wù)5執(zhí)行任務(wù)完成
任務(wù)9執(zhí)行任務(wù)完成
任務(wù)3執(zhí)行任務(wù)完成
任務(wù)1執(zhí)行任務(wù)完成
任務(wù)8執(zhí)行任務(wù)完成
Thread-6獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:10
Thread-0獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:10
Thread-7獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:10
Thread-2獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:10
Thread-4獲取令牌成功，執(zhí)行時(shí)間:2025-07-02 09:47:10
任務(wù)0執(zhí)行任務(wù)完成
任務(wù)4執(zhí)行任務(wù)完成
任務(wù)6執(zhí)行任務(wù)完成
任務(wù)7執(zhí)行任務(wù)完成
任務(wù)2執(zhí)行任務(wù)完成三、流限線程池優(yōu)化思路
當(dāng)然關(guān)于流限，如果用戶明確知曉超時(shí)中斷的時(shí)機(jī)并具備靈活響應(yīng)中斷的能力，我們完全可以補(bǔ)充一個(gè)帶有超時(shí)限制的任務(wù)提交函數(shù)trySubmitTask：
public boolean trySubmitTask(final Runnable command,
long timeout,
TimeUnit unit) throws InterruptedException {
if (!semaphore.tryAcquire(timeout, unit)) {
Console.log("{}獲取令牌失敗，執(zhí)行時(shí)間:{}", Thread.currentThread().getName(), DateUtil.now());
return false;
}
Console.log("{}獲取令牌成功，執(zhí)行時(shí)間:{}", Thread.currentThread().getName(), DateUtil.now());
try {
threadPool.execute(() -> {
try {
//執(zhí)行任務(wù)
command.run();
} finally {
//線程執(zhí)行完成后釋放令牌
semaphore.release();
}
});
} catch (RejectedExecutionException e) {//異常兜底
semaphore.release();
return false;
}
return true;
}對(duì)應(yīng)我們也給出使用示例：
public static void main(String[] args) {
RateLimitedExecutor executor = new RateLimitedExecutor(1);
for (int i = 0; i < 2; i++) {
new Thread(new Task("任務(wù)" + i, executor)).start();
}
}
private static class Task implements Runnable {
private final String threadName;
private final RateLimitedExecutor executor;
public Task(String threadName, RateLimitedExecutor executor) {
this.threadName = threadName;
this.executor = executor;
}
@SneakyThrows
@Override
public void run() {
executor.trySubmitTask(() -> {
ThreadUtil.sleep(5000);
Console.log("{}執(zhí)行任務(wù)完成", threadName);
}, 1, TimeUnit.SECONDS);
}
}輸出結(jié)果如下，可以看到線程1按照正確的超時(shí)等待返回了：
Thread-0獲取令牌成功，執(zhí)行時(shí)間:2025-07-04 09:28:27
Thread-1獲取令牌失敗，執(zhí)行時(shí)間:2025-07-04 09:28:28
任務(wù)0執(zhí)行任務(wù)完成四、小結(jié)
針對(duì)線程池拒絕策略的設(shè)計(jì)和使用更多是考察讀者對(duì)于線程池源碼的理解和使用經(jīng)驗(yàn)，這里筆者僅在思路上給出示例，當(dāng)然實(shí)現(xiàn)上也存在很多不完美的地方，例如：
- 如何保證持久化任務(wù)被可靠消費(fèi)。
- 如何保證數(shù)據(jù)庫(kù)和內(nèi)存中任務(wù)的公平調(diào)度。
- 持久化任務(wù)是先刪后返回還是先返回處理完成后刪除如何決定？
