線程池管理工具類

 

各位讀者，大家好！

   

   我們在項目開發過程中，經常會使用線程池管理，特別是對TPS有一定要求的情況。

   線程池會自動幫助我們管理線程的創建、回收及銷毀等工作，方便我們的開發。特別有異步需求、調用第三方系統且比較耗時的批量任務，都可以使用線程池很好的幫我們節省大量時間，提高用戶的體驗。

 

   比如：小美想從XX系統進行貸款，那么XX系統肯定需要獲取小美的人行征信，還可能需要校驗小美是否從此種同類的系統已經貸款，如果貸款了，那么就不能再次貸款。這里牽扯到2個第三方系統，人行征信系統 及 貸款信息匯總系統，假設從人行征信系統獲取征信需要5s，取貸款信息匯總系統需要3s，那么按照正常的同步流程：先獲取征信，再獲取貸款信息，就一共需要8s；那么如果再加上其他各種各樣的校驗、第三方外調（甚至外調超時），那么加起來時間更長，對客戶的體驗肯定會受影響，可能就會丟失小美這樣大量的客戶。

    所以，我們可以使用線程池來做這件事情，讓A線程去調人行征信，讓B線程去獲取貸款信息，讓C線程去調其他第三方依賴系統....,這樣，一旦任何一個線程得到不符合貸款的校驗結果，就返回並通知終止其他線程。這樣，客戶等待的時間就成倍節約下來，理論上最好情況是等待了花費最長的那個線程的時間。不僅如此，假設所有線程校驗都通過，我們需要等待以上所有線程執行完畢，才可以繼續接下來的工作，這個時候我們就可以用到本次介紹的工具類中awaitTasksFinished方法即可。

    以下是該線程管理工具的完整代碼：

 

package com.cheng2839.utils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 線程池工具類
 *
 * @author cheng2839
 * @date 2018年11月16日
 */
public final class ThreadPoolFactory {

    private static final Logger logger = LoggerFactory.getLogger(ThreadPoolFactory.class);

    private static ThreadPoolFactory factory = new ThreadPoolFactory();

    //線程池默認配置設置
    private static final int CORE_POOL_SIZE = 5;
    private static final int MAXIMUM_POOL_SIZE = 20;
    private static final long KEEP_ALIVE_TIME = 60L;

    private ThreadPoolFactory() {
    }

    public static ThreadPoolFactory getFactory() {
        if (factory == null) {
            factory = new ThreadPoolFactory();
        }
        return factory;
    }

    /**
     * 創建一個默認的線程池
     *
     * @return
     * @author cheng2839
     * @date 2018年11月16日
     */
    public ExecutorService getDefaultThreadPool() {
        //配置最大隊列容量
        BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<>(Integer.MAX_VALUE);
        return getCustomThreadPool(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME, blockingQueue);
    }

    /**
     * 創建一個簡單的線程池
     *
     * @return
     * @author cheng2839
     * @date 2018年11月16日
     */
    public ExecutorService getSimpleThreadPool(int corePoolSize, int maximumPoolSize) {
        return getCustomThreadPool(corePoolSize, maximumPoolSize, KEEP_ALIVE_TIME, new LinkedBlockingQueue<Runnable>());
    }

    /**
     * 創建一個指定隊列的線程池
     *
     * @return
     * @author cheng2839
     * @date 2018年11月16日
     */
    public ExecutorService getCustomQueueThreadPool(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> blockingQueue) {
        return getCustomThreadPool(corePoolSize, maximumPoolSize, KEEP_ALIVE_TIME, blockingQueue);
    }

    /**
     * 創建可跟蹤任務狀態的執行器
     *
     * @return
     * @author cheng2839
     * @date 2018年11月16日
     */
    public ExecutorCompletionService getCompletionService(ExecutorService executorService) {
        return new ExecutorCompletionService(executorService);
    }

    /**
     * 創建一個定制化的線程池
     *
     * @return
     * @author cheng2839
     * @date 2018年11月16日
     */
    public ExecutorService getCustomThreadPool(int corePoolSize,
                                               int maximumPoolSize,
                                               Long keepAliveTime,
                                               BlockingQueue<Runnable> blockingQueue) {
        logger.info("開始初始化線程池[corePoolSize={},maximumPoolSize={},keepAliveTime={}s]...", corePoolSize, maximumPoolSize, keepAliveTime);

        RejectedExecutionHandler rejectedExecutionHandler = (Runnable r, ThreadPoolExecutor executor) -> {
            logger.error("線程池已滿，任務被丟棄...........................");
            return;
        };
        ExecutorService executorService = new ThreadPoolExecutor(corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                TimeUnit.SECONDS,
                blockingQueue,
                rejectedExecutionHandler);
        logger.info("初始化線程池完成！");
        return executorService;
    }

    /**
     * 等待任務執行完成 並 釋放連接池
     *
     * @param futureList
     * @param completionService
     * @param <V>
     * @author cheng2839
     * @date 2018年11月16日
     */
    public <V> void awaitTasksFinished(List<Future> futureList, CompletionService<V> completionService) {
        try {
            if (!CollectionUtils.isEmpty(futureList) && completionService != null) {
                logger.info("等待批量任務[{}]執行。。。", futureList.size());
                for (int n = 0; n < futureList.size(); n++) {
                    Future future = completionService.take();
                    if (future != null)
                        future.get();
                }
            }
        } catch (Exception e) {
            logger.error("多線程獲取結果異常: {}", e);
        }
    }

    /**
     * 關閉線程池
     *
     * @param executorService
     * @author cheng2839
     * @date 2018年11月16日
     */
    public void shutdown(ExecutorService executorService) {
        try {
            if (executorService != null) {
                logger.info("關閉線程池:{}", executorService);
                executorService.shutdown();
            }
        } catch (Exception e) {
            if (!executorService.isTerminated()) {
                executorService.shutdownNow();
            }
        } finally {
            try {
                if (executorService != null && !executorService.isShutdown()) {
                    executorService.shutdown();
                }
            } catch (Exception e) {
                logger.error("線程池關閉異常：{}", e);
            }
        }
    }

}

 

    運行示例略，大家可以根據自己的需求靈活運用！

 

    源碼歸本人所有，如需轉載，請標注出處。

 

我還是忍不住放上了這張圖片，萬一你想和我說說話呢😄❀