SpringBoot開發案例之多任務並行+線程池處理

前言

前幾篇文章着重介紹了後端服務數據庫和多線程並行處理優化，並示例了改造先後的僞代碼邏輯。固然了，優化是無止境的，前人栽樹後人乘涼。做爲咱們開發者來講，既然站在了巨人的肩膀上，就要寫出更加優化的程序。

• SpringBoot開發案例之JdbcTemplate批量操做

• SpringBoot開發案例之CountDownLatch多任務並行處理

改造

理論上講，線程越多程序可能更快，可是在實際使用中咱們須要考慮到線程自己的建立以及銷燬的資源消耗，以及保護操做系統自己的目的。咱們一般須要將線程限制在必定的範圍之類，線程池就起到了這樣的做用。

程序邏輯

一張圖能解決的問題，就應該儘量的少BB，固然底層原理性的東西仍是須要你們去記憶並理解的。

Java 線程池

Java經過Executors提供四種線程池，分別爲：

• newCachedThreadPool建立一個可緩存線程池，若是線程池長度超過處理須要，可靈活回收空閒線程，若無可回收，則新建線程。
• newFixedThreadPool 建立一個定長線程池，可控制線程最大併發數，超出的線程會在隊列中等待。
• newScheduledThreadPool 建立一個定長線程池，支持定時及週期性任務執行。
• newSingleThreadExecutor 建立一個單線程化的線程池，它只會用惟一的工做線程來執行任務，保證全部任務按照指定順序(FIFO, LIFO, 優先級)執行。

優勢

• 重用存在的線程，減小對象建立、消亡的開銷，性能佳。
• 可有效控制最大併發線程數，提升系統資源的使用率，同時避免過多資源競爭，避免堵塞。
• 提供定時執行、按期執行、單線程、併發數控制等功能。

代碼實現

方式一(CountDownLatch)

/**
 * 多任務並行+線程池統計
 * 建立者 科幫網  https://blog.52itstyle.com
 * 建立時間    2018年4月17日
 */
public class StatsDemo {
    final static SimpleDateFormat sdf = new SimpleDateFormat(
            "yyyy-MM-dd HH:mm:ss");
    
    final static String startTime = sdf.format(new Date());
    
    /**
     * IO密集型任務  = 通常爲2*CPU核心數（常出現於線程中：數據庫數據交互、文件上傳下載、網絡數據傳輸等等）
     * CPU密集型任務 = 通常爲CPU核心數+1（常出現於線程中：複雜算法）
     * 混合型任務  = 視機器配置和複雜度自測而定
     */
    private static int corePoolSize = Runtime.getRuntime().availableProcessors();
    /**
     * public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
     *                           TimeUnit unit,BlockingQueue<Runnable> workQueue)
     * corePoolSize用於指定核心線程數量
     * maximumPoolSize指定最大線程數
     * keepAliveTime和TimeUnit指定線程空閒後的最大存活時間
     * workQueue則是線程池的緩衝隊列,還未執行的線程會在隊列中等待
     * 監控隊列長度，確保隊列有界
     * 不當的線程池大小會使得處理速度變慢，穩定性降低，而且致使內存泄露。若是配置的線程過少，則隊列會持續變大，消耗過多內存。
     * 而過多的線程又會 因爲頻繁的上下文切換致使整個系統的速度變緩——殊途而同歸。隊列的長度相當重要，它必須得是有界的，這樣若是線程池不堪重負了它能夠暫時拒絕掉新的請求。
     * ExecutorService 默認的實現是一個無界的 LinkedBlockingQueue。
     */
    private static ThreadPoolExecutor executor  = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(1000));
    
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(5);
        //使用execute方法
        executor.execute(new Stats("任務A", 1000, latch));
        executor.execute(new Stats("任務B", 1000, latch));
        executor.execute(new Stats("任務C", 1000, latch));
        executor.execute(new Stats("任務D", 1000, latch));
        executor.execute(new Stats("任務E", 1000, latch));
        latch.await();// 等待全部人任務結束
        System.out.println("全部的統計任務執行完成:" + sdf.format(new Date()));
    }

    static class Stats implements Runnable  {
        String statsName;
        int runTime;
        CountDownLatch latch;

        public Stats(String statsName, int runTime, CountDownLatch latch) {
            this.statsName = statsName;
            this.runTime = runTime;
            this.latch = latch;
        }

        public void run() {
            try {
                System.out.println(statsName+ " do stats begin at "+ startTime);
                //模擬任務執行時間
                Thread.sleep(runTime);
                System.out.println(statsName + " do stats complete at "+ sdf.format(new Date()));
                latch.countDown();//單次任務結束，計數器減一
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

方式二（Future）

/**
 * 多任務並行+線程池統計
 * 建立者 科幫網 https://blog.52itstyle.com
 * 建立時間    2018年4月17日
 */
public class StatsDemo {
    final static SimpleDateFormat sdf = new SimpleDateFormat(
            "yyyy-MM-dd HH:mm:ss");
    
    final static String startTime = sdf.format(new Date());
    
    /**
     * IO密集型任務  = 通常爲2*CPU核心數（常出現於線程中：數據庫數據交互、文件上傳下載、網絡數據傳輸等等）
     * CPU密集型任務 = 通常爲CPU核心數+1（常出現於線程中：複雜算法）
     * 混合型任務  = 視機器配置和複雜度自測而定
     */
    private static int corePoolSize = Runtime.getRuntime().availableProcessors();
    /**
     * public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
     *                           TimeUnit unit,BlockingQueue<Runnable> workQueue)
     * corePoolSize用於指定核心線程數量
     * maximumPoolSize指定最大線程數
     * keepAliveTime和TimeUnit指定線程空閒後的最大存活時間
     * workQueue則是線程池的緩衝隊列,還未執行的線程會在隊列中等待
     * 監控隊列長度，確保隊列有界
     * 不當的線程池大小會使得處理速度變慢，穩定性降低，而且致使內存泄露。若是配置的線程過少，則隊列會持續變大，消耗過多內存。
     * 而過多的線程又會 因爲頻繁的上下文切換致使整個系統的速度變緩——殊途而同歸。隊列的長度相當重要，它必須得是有界的，這樣若是線程池不堪重負了它能夠暫時拒絕掉新的請求。
     * ExecutorService 默認的實現是一個無界的 LinkedBlockingQueue。
     */
    private static ThreadPoolExecutor executor  = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(1000));
    
    public static void main(String[] args) throws InterruptedException {
        List<Future<String>> resultList = new ArrayList<Future<String>>(); 
        //使用submit提交異步任務，而且獲取返回值爲future
        resultList.add(executor.submit(new Stats("任務A", 1000)));
        resultList.add(executor.submit(new Stats("任務B", 1000)));
        resultList.add(executor.submit(new Stats("任務C", 1000)));
        resultList.add(executor.submit(new Stats("任務D", 1000)));
        resultList.add(executor.submit(new Stats("任務E", 1000)));
       //遍歷任務的結果
        for (Future<String> fs : resultList) { 
            try { 
                System.out.println(fs.get());//打印各個線任務執行的結果，調用future.get() 阻塞主線程，獲取異步任務的返回結果
            } catch (InterruptedException e) { 
                e.printStackTrace(); 
            } catch (ExecutionException e) { 
                e.printStackTrace(); 
            } finally { 
                //啓動一次順序關閉，執行之前提交的任務，但不接受新任務。若是已經關閉，則調用沒有其餘做用。
                executor.shutdown(); 
            } 
        } 
        System.out.println("全部的統計任務執行完成:" + sdf.format(new Date()));
    }

    static class Stats implements Callable<String>  {
        String statsName;
        int runTime;

        public Stats(String statsName, int runTime) {
            this.statsName = statsName;
            this.runTime = runTime;
        }

        public String call() {
            try {
                System.out.println(statsName+ " do stats begin at "+ startTime);
                //模擬任務執行時間
                Thread.sleep(runTime);
                System.out.println(statsName + " do stats complete at "+ sdf.format(new Date()));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return call();
        }
    }
}

執行時間

以上代碼，均是僞代碼，下面是2000+個學生的真實測試記錄。

2018-04-17 17:42:29.284 INFO   測試記錄81e51ab031eb4ada92743ddf66528d82-單線程順序執行，花費時間:3797
2018-04-17 17:42:31.452 INFO   測試記錄81e51ab031eb4ada92743ddf66528d82-多線程並行任務，花費時間:2167
2018-04-17 17:42:33.170 INFO   測試記錄81e51ab031eb4ada92743ddf66528d82-多線程並行任務+線程池，花費時間:1717