快捷導(dǎo)航

關(guān)于線程池異步線程中再次獲取線程池資源的問題

更新時間：2024年08月20日 10:33:56 作者：BlueKitty1210

這篇文章主要介紹了關(guān)于線程池異步線程中再次獲取線程池資源的問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

問題描述

在線上發(fā)生的一次問題，在場景中有這樣一個業(yè)務(wù)，需要異步執(zhí)行一個主任務(wù)，主任務(wù)中又包含著N個子任務(wù)，為了整個主任務(wù)能夠快速處理，又將子任務(wù)按照數(shù)量獲取線程資源異步處理，即異步線程A中再異步調(diào)用A1，A2，A3. A可能同時存在多個.

實際場景中，由于系統(tǒng)線程池分配數(shù)量較小，且一段時間內(nèi)先后啟動了多個主任務(wù)，耗時的主任務(wù)中又用子任務(wù)取申請線程導(dǎo)致線程池資源耗盡

問題原因

1. 主任務(wù)是從線程池中獲取的線程資源，同時主任務(wù)比較耗時?

2. 每個主任務(wù)中包含的N的子任務(wù)，會再申請線程，處理完畢釋放回線程池

3. 啟動了多個主任務(wù)時，每個主任務(wù)在未結(jié)束之前，都會占用自身一個線程不會釋放，消耗一個線程池資源

4. 后期頻繁啟動主任務(wù)，可能使數(shù)量=線程池線程數(shù)，此時子任務(wù)無法再從線程池獲得資源，就進(jìn)入隊列等待

5. 最終結(jié)果就造成了每個主任務(wù)都占用線程，但主任務(wù)內(nèi)的子任務(wù)無法獲取線程，線程池癱瘓不可用

問題復(fù)現(xiàn)

package test;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @title 線程池異步線程中再次獲取線程池資源的問題
 * @author Xingbz
 * @description
 *  記;
 *
 *  究其原因在于:
 *      
 * @createDate 2020-7-17
 */
@Slf4j
public class TestWork {

    private static final ThreadPoolTaskExecutor EXECUTOR;

    static {
        EXECUTOR = myExecutor();
    }

    /** 初始化線程池 */
    public static ThreadPoolTaskExecutor myExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心線程數(shù)
        executor.setCorePoolSize(5);
        // 最大線程數(shù)
        executor.setMaxPoolSize(20);
        // 排隊任務(wù)隊列
        executor.setQueueCapacity(100);
        // 線程名稱前綴
        executor.setThreadNamePrefix("異步線程-");
        // 隊列滿后拒絕策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 線程最大回收時間
        executor.setKeepAliveSeconds(100);
        // 初始化線程
        executor.initialize();
        return executor;
    }

    /** 模擬測試 */
    public static void main(String[] args) throws Exception {
        // 主任務(wù)數(shù)量
        int mainJobNum = 20;

        CountDownLatch mainDownLatch = new CountDownLatch(mainJobNum);

        for (int i = 0; i < mainJobNum; i++) {
            // 主任務(wù)編號, 方便區(qū)分
            int index = i + 1;

            // 模擬每1秒開始一個主任務(wù)
            TimeUnit.SECONDS.sleep(1);

            EXECUTOR.submit(() -> {
                try {
                    log.debug("\t執(zhí)行主任務(wù)" + index);

                    // 每個主任務(wù)隨機(jī)包含N個子任務(wù), 再異步調(diào)用線程池資源處理
                    int subJobNum = RandomUtils.nextInt(2, 3);
                    subJobWorkAsync(subJobNum, index);
                } finally {
                    mainDownLatch.countDown();
                }
            });
        }

        mainDownLatch.await();
        EXECUTOR.shutdown();
        log.info("完成所有任務(wù) > > >");
    }

    /** 異步執(zhí)行子任務(wù) */
    private static void subJobWorkAsync(int subJobNum, int index) {
        CountDownLatch subDownLatch = new CountDownLatch(subJobNum);
        for (int j = 0; j < subJobNum; j++) {
            EXECUTOR.submit(() -> {
                try {
                    log.warn("\t\t\t執(zhí)行一個" + index + "的子任務(wù)");
                    // 每個子任務(wù)模擬耗時
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    subDownLatch.countDown();
                }
            });
        }

        try {
            subDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

執(zhí)行代碼，結(jié)果如下：