SpringBoot 多任務並行+線程池處理的實現

前言

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

SpringBoot開發案例之JdbcTemplate批量操作
SpringBoot開發案例之CountDownLatch多任務並行處理

改造

理論上講，線程越多程序可能更快，但是在實際使用中我們需要考慮到線程本身的創建以及銷毀的資源消耗，以及保護操作系統本身的目的。我們通常需要將線程限制在一定的範圍之類，線程池就起到了這樣的作用。

程序邏輯

多任務並行+線程池處理.png

一張圖能解決的問題，就應該盡可能的少BB，當然底層原理性的東西還是需要大家去記憶並理解的。

Java 線程池

Java通過Executors提供四種線程池，分別為：

1. newCachedThreadPool創建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閒線程，若無可回收，則新建線程。
2. newFixedThreadPool 創建一個定長線程池，可控制線程最大並發數，超出的線程會在隊列中等待。
3. newScheduledThreadPool 創建一個定長線程池，支持定時及週期性任務執行。
4. newSingleThreadExecutor 創建一個單線程化的線程池，它只會用唯一的工作線程來執行任務，保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行。

優點

1. 重用存在的線程，減少對象創建、消亡的開銷，性能佳。
2. 可有效控制最大並發線程數，提高系統資源的使用率，同時避免過多資源競爭，避免堵塞。
3. 提供定時執行、定期執行、單線程、並發數控制等功能。

代碼實現

方式一(CountDownLatch)

 /** * 多任務並行+線程池統計* 創建時間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）

 /** * 多任務並行+線程池統計* 創建時間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

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持武林網。