不同的業(yè)務(wù)場(chǎng)景對(duì)任務(wù)丟失的容忍度、響應(yīng)延遲的要求、系統(tǒng)保護(hù)的需求各不相同。下面通過(guò) 6 個(gè)典型場(chǎng)景，分析如何選擇合適的拒絕策略，并給出代碼示例和注意事項(xiàng)。

場(chǎng)景1：電商訂單支付（核心交易鏈路）

業(yè)務(wù)特點(diǎn)：

• 每一筆支付請(qǐng)求都必須處理，不能丟失。
• 支付操作涉及數(shù)據(jù)庫(kù)更新、第三方網(wǎng)關(guān)調(diào)用、消息發(fā)送等。
• 要求高一致性，失敗需要明確感知并觸發(fā)重試或補(bǔ)償。

壓力情況：大促時(shí)瞬間流量激增，線程池可能飽和。

選擇策略：AbortPolicy + 上層統(tǒng)一捕獲異常，進(jìn)行異步重試或放入死信隊(duì)列。

理由：

• 支付任務(wù)絕對(duì)不能靜默丟棄（DiscardPolicy 不可用）。
• 不能讓調(diào)用者線程執(zhí)行支付任務(wù)（CallerRunsPolicy 會(huì)阻塞 Tomcat 線程，導(dǎo)致整個(gè)服務(wù)響應(yīng)變慢）。
• 拋出異常是最明確的失敗信號(hào)，由調(diào)用方（通常是 Controller 或 Service）捕獲后，可以將任務(wù)轉(zhuǎn)儲(chǔ)到消息隊(duì)列或數(shù)據(jù)庫(kù)，稍后重試。

代碼示例：

// 線程池配置
@Bean("paymentExecutor")
public Executor paymentExecutor() {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
        20, 50, 60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(200),
        new NamedThreadFactory("payment"),
        new ThreadPoolExecutor.AbortPolicy()  // 顯式拋出異常
    );
    return executor;
}

// 業(yè)務(wù)調(diào)用處
@Service
public class PaymentService {
    @Autowired
    private ThreadPoolExecutor paymentExecutor;
    
    public void processPayment(PaymentRequest request) {
        try {
            paymentExecutor.execute(() -> doPayment(request));
        } catch (RejectedExecutionException e) {
            // 線程池飽和，將任務(wù)寫(xiě)入重試隊(duì)列（如 RocketMQ、Redis 等）
            saveToRetryQueue(request);
            log.warn("Payment task rejected, saved to retry queue. requestId={}", request.getId());
            // 可選：向調(diào)用方返回“系統(tǒng)繁忙，稍后重試”的提示
        }
    }
}

監(jiān)控指標(biāo)：拒絕次數(shù)必須為 0，一旦出現(xiàn)立即告警并擴(kuò)容。

場(chǎng)景2：秒殺扣庫(kù)存（瞬時(shí)高并發(fā)，允許快速失敗）

業(yè)務(wù)特點(diǎn)：

• 請(qǐng)求量瞬間爆炸，但真正能成功秒殺到的用戶只有一小部分。
• 要求極低的響應(yīng)延遲，不能排隊(duì)等待。
• 超出處理能力的請(qǐng)求應(yīng)該被快速拒絕，返回“已售罄”或“系統(tǒng)繁忙”。

壓力情況：QPS 從幾百瞬間飆升到幾十萬(wàn)。

選擇策略：AbortPolicy 或 DiscardPolicy + 前端友好提示。

理由：

• 使用 SynchronousQueue + 有限最大線程數(shù)（如 200），任何超出并發(fā)能力的請(qǐng)求立即觸發(fā)拒絕。
• 用 AbortPolicy 拋異常，或者用 DiscardPolicy 靜默丟棄，但都需要在業(yè)務(wù)層捕獲并返回統(tǒng)一的失敗響應(yīng)。
• 不能使用 CallerRunsPolicy，因?yàn)檎{(diào)用者（Tomcat 工作線程）執(zhí)行秒殺任務(wù)會(huì)嚴(yán)重拖垮整個(gè) Web 容器。

代碼示例：

// 秒殺專(zhuān)用線程池
int maxConcurrency = 200; // 根據(jù)壓測(cè)得出系統(tǒng)能承受的最大并發(fā)扣庫(kù)存操作
ExecutorService seckillExecutor = new ThreadPoolExecutor(
    0, maxConcurrency, 30L, TimeUnit.SECONDS,
    new SynchronousQueue<>(),
    new NamedThreadFactory("seckill"),
    new ThreadPoolExecutor.AbortPolicy()
);
// 秒殺接口
@PostMapping("/seckill")
public Result seckill(Long goodsId, Long userId) {
    try {
        seckillExecutor.execute(() -> {
            // 扣庫(kù)存、創(chuàng)建訂單等核心操作
            inventoryService.decr(goodsId);
            orderService.create(userId, goodsId);
        });
        return Result.success("搶購(gòu)中，請(qǐng)稍后查看訂單");
    } catch (RejectedExecutionException e) {
        // 線程池滿，直接返回失敗
        return Result.error("很遺憾，您沒(méi)搶到，下次加油");
    }
}

優(yōu)化點(diǎn)：可以在拒絕策略中直接記錄指標(biāo)，但無(wú)需重試，因?yàn)槊霘⑹【褪亲罱K結(jié)果。

場(chǎng)景3：異步發(fā)送短信/郵件（非關(guān)鍵通知，允許少量丟失）

業(yè)務(wù)特點(diǎn)：

• 用戶注冊(cè)、下單后發(fā)送確認(rèn)短信/郵件。
• 即使少量消息發(fā)送失敗，也不影響核心業(yè)務(wù)（用戶可以通過(guò)其他渠道重試）。
• 不希望因消息發(fā)送阻塞主流程。

壓力情況：業(yè)務(wù)高峰時(shí)消息量較大，但系統(tǒng)可以接受一定程度的丟棄。

選擇策略：DiscardOldestPolicy 或 DiscardPolicy + 日志記錄。

理由：

• 消息堆積過(guò)久反而不如丟棄舊消息，保證新消息能及時(shí)發(fā)出（DiscardOldestPolicy）。
• 如果消息完全可丟棄（如運(yùn)營(yíng)推廣短信），直接用 DiscardPolicy。
• 不能使用 CallerRunsPolicy，因?yàn)橹骶€程（如訂單完成后的異步通知）不應(yīng)被阻塞。

代碼示例：

// 通知線程池
ThreadPoolExecutor notifyExecutor = new ThreadPoolExecutor(
    5, 20, 60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(500),
    new NamedThreadFactory("notify"),
    new ThreadPoolExecutor.DiscardOldestPolicy()
);
// 發(fā)送短信
public void sendSms(String phone, String content) {
    notifyExecutor.execute(() -> {
        try {
            smsClient.send(phone, content);
        } catch (Exception e) {
            log.error("Send sms failed, phone={}", phone, e);
            // 可選：記錄失敗到數(shù)據(jù)庫(kù)，由定時(shí)任務(wù)補(bǔ)償
        }
    });
}

監(jiān)控：可以統(tǒng)計(jì)丟棄數(shù)量，如果丟棄率過(guò)高（如 >1%），考慮擴(kuò)容或優(yōu)化短信通道。

場(chǎng)景4：日志/審計(jì)記錄（海量低價(jià)值，可丟棄）

業(yè)務(wù)特點(diǎn)：

• 每條請(qǐng)求都需要記錄訪問(wèn)日志、用戶行為日志。
• 數(shù)據(jù)量極大（每秒數(shù)萬(wàn)條），對(duì)實(shí)時(shí)性要求低。
• 偶爾丟失幾條日志對(duì)業(yè)務(wù)無(wú)影響。

壓力情況：持續(xù)高吞吐，磁盤(pán)或網(wǎng)絡(luò)可能成為瓶頸。

選擇策略：DiscardPolicy（靜默丟棄）。

理由：

• 日志系統(tǒng)不應(yīng)該拖垮主業(yè)務(wù)。如果線程池滿了，說(shuō)明下游（如日志服務(wù)器、ES）已經(jīng)處理不過(guò)來(lái)，再排隊(duì)只會(huì)加劇問(wèn)題。
• 直接丟棄是最簡(jiǎn)單有效的自我保護(hù)。
• 也可以自定義策略，將丟棄的日志采樣記錄（用于分析丟失率）。

代碼示例：

ThreadPoolExecutor logExecutor = new ThreadPoolExecutor(
    2, 10, 10L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(1000),
    new NamedThreadFactory("access-log"),
    new ThreadPoolExecutor.DiscardPolicy()
);
// 記錄訪問(wèn)日志
public void logAccess(HttpServletRequest request) {
    logExecutor.execute(() -> {
        // 構(gòu)建日志對(duì)象，發(fā)送到 Kafka 或?qū)懭氡镜匚募?
        accessLogService.save(parseLog(request));
    });
}

進(jìn)階：可以結(jié)合采樣，在丟棄時(shí)隨機(jī)記錄 1% 的丟棄事件用于監(jiān)控。

場(chǎng)景5：批量數(shù)據(jù)導(dǎo)入（任務(wù)重，不允許丟失，可接受延遲）

業(yè)務(wù)特點(diǎn)：

• 從文件、數(shù)據(jù)庫(kù)批量導(dǎo)入數(shù)據(jù)，每個(gè)任務(wù)執(zhí)行時(shí)間較長(zhǎng)（秒級(jí)到分鐘級(jí)）。
• 任務(wù)數(shù)量固定，不允許丟失。
• 可以接受導(dǎo)入速度變慢，但不能失敗。

壓力情況：任務(wù)提交可能短時(shí)間集中，但總?cè)蝿?wù)量可控。

選擇策略：CallerRunsPolicy。

理由：

• 當(dāng)線程池飽和時(shí)，由調(diào)用者線程（例如主線程或定時(shí)任務(wù)線程）直接執(zhí)行導(dǎo)入任務(wù)，這樣不會(huì)丟失任務(wù)，同時(shí)會(huì)自然降低新任務(wù)的提交速度。
• 因?yàn)閷?dǎo)入任務(wù)本身是重量級(jí)操作，調(diào)用者執(zhí)行雖然會(huì)阻塞，但總比丟棄好。
• 配合有界隊(duì)列，防止內(nèi)存溢出。

代碼示例：

ThreadPoolExecutor importExecutor = new ThreadPoolExecutor(
    4, 8, 5L, TimeUnit.MINUTES,
    new ArrayBlockingQueue<>(10),  // 小隊(duì)列，讓拒絕策略盡快生效
    new NamedThreadFactory("data-import"),
    new ThreadPoolExecutor.CallerRunsPolicy()
);
// 批量提交導(dǎo)入任務(wù)
public void importLargeFiles(List<File> files) {
    for (File file : files) {
        importExecutor.execute(() -> importOneFile(file));
    }
    importExecutor.shutdown();
    importExecutor.awaitTermination(1, TimeUnit.HOURS);
}

注意：CallerRunsPolicy 可能會(huì)導(dǎo)致調(diào)用者線程長(zhǎng)時(shí)間阻塞，如果調(diào)用者是定時(shí)任務(wù)線程，可能影響其他定時(shí)任務(wù)?？梢詫⒄{(diào)用者線程池也設(shè)置得足夠健壯。

場(chǎng)景6：與消息隊(duì)列結(jié)合（最終一致性，高可靠性）

業(yè)務(wù)特點(diǎn)：

• 任務(wù)必須處理，但不能阻塞當(dāng)前線程。
• 希望削峰填谷，利用消息隊(duì)列的持久化能力。

選擇策略：自定義拒絕策略，將任務(wù)轉(zhuǎn)發(fā)到 RocketMQ、Kafka 等。

理由：

• 線程池只處理實(shí)時(shí)部分，當(dāng)線程池飽和時(shí)，將任務(wù)寫(xiě)入消息隊(duì)列，由消費(fèi)者異步處理。
• 這樣既保護(hù)了系統(tǒng)，又保證了任務(wù)不丟失。

代碼示例：

public class MQBackedRejectedHandler implements RejectedExecutionHandler {
    private final RocketMQTemplate mqTemplate;
    private final String topic;
    public MQBackedRejectedHandler(RocketMQTemplate mqTemplate, String topic) {
        this.mqTemplate = mqTemplate;
        this.topic = topic;
    }
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (executor.isShutdown()) {
            return;
        }
        // 將任務(wù)序列化后發(fā)送到 MQ
        if (r instanceof SerializableTask) {
            mqTemplate.syncSend(topic, ((SerializableTask) r).getPayload());
        } else {
            // 兜底：記錄到數(shù)據(jù)庫(kù)
            saveToDatabase(r);
        }
    }
}
// 線程池配置
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10, 50, 60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(100),
    new NamedThreadFactory("worker"),
    new MQBackedRejectedHandler(mqTemplate, "rejected-task-topic")
);

注意：發(fā)送 MQ 本身也可能失敗，需要做好重試和監(jiān)控。

場(chǎng)景總結(jié)表

最佳實(shí)踐建議

1. 默認(rèn)不要使用 AbortPolicy 而毫無(wú)處理：至少要在業(yè)務(wù)代碼中捕獲 RejectedExecutionException，記錄日志或觸發(fā)降級(jí)。
2. 非核心業(yè)務(wù)優(yōu)先使用 DiscardPolicy 并記錄丟棄次數(shù)：用于容量規(guī)劃。
3. 所有拒絕策略都應(yīng)該有監(jiān)控：通過(guò) Micrometer、Prometheus 暴露 rejected.count 指標(biāo)。
4. CallerRunsPolicy 要謹(jǐn)慎評(píng)估調(diào)用者線程：如果調(diào)用者是 Web 請(qǐng)求線程，可能導(dǎo)致請(qǐng)求超時(shí)堆積。
5. 自定義拒絕策略時(shí)不要執(zhí)行過(guò)于耗時(shí)的操作（如寫(xiě)數(shù)據(jù)庫(kù)、發(fā) MQ），否則會(huì)加劇線程池的阻塞。

通過(guò)結(jié)合具體業(yè)務(wù)場(chǎng)景選擇合適的拒絕策略，可以平衡系統(tǒng)穩(wěn)定性、任務(wù)可靠性和響應(yīng)延遲三者之間的關(guān)系，構(gòu)建高可用的并發(fā)系統(tǒng)。

到此這篇關(guān)于Java線程池拒絕策略場(chǎng)景分析的文章就介紹到這了,更多相關(guān)Java線程池拒絕策略內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論