建站
咨詢
售后
建站咨詢
新聞中心
前言
在我們的日常的編程當中，并發(fā)是始終離不開的主題，而在并發(fā)多線程當中，線程池又是一個不可規(guī)避的問題。多線程可以提高我們并發(fā)程序的效率，可以讓我們不去頻繁的申請和釋放線程，這是一個很大的花銷，而在線程池當中就不需要去頻繁的申請線程，他的主要原理是申請完線程之后并不中斷，而是不斷的去隊列當中領取任務，然后執(zhí)行，反復這樣的操作。在本篇文章當中我們主要是介紹線程池的原理，因此我們會自己寫一個非常非常簡單的線程池，主要幫助大家理解線程池的核心原理?。。?/p>
臨武網站制作公司哪家好，找成都創(chuàng)新互聯(lián)公司！從網頁設計、網站建設、微信開發(fā)、APP開發(fā)、成都響應式網站建設公司等網站項目制作，到程序開發(fā)，運營維護。成都創(chuàng)新互聯(lián)公司2013年至今到現(xiàn)在10年的時間，我們擁有了豐富的建站經驗和運維經驗，來保證我們的工作的順利進行。專注于網站建設就選成都創(chuàng)新互聯(lián)公司。
線程池給我們提供的功能
我們首先來看一個使用線程池的例子：
import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors; public class Demo01 { public static void main(String[] args) { ExecutorService pool = Executors.newFixedThreadPool(5); for (int i = 0; i < 100; i++) { pool.execute(new Runnable() { @Override public void run() { for (int i = 0; i < 100; i++) { System.out.println(Thread.currentThread().getName() + " print " + i); } } }); } }}在上面的例子當中，我們使用Executors.newFixedThreadPool去生成來一個固定線程數(shù)目的線程池，在上面的代碼當中我們是使用5個線程，然后通過execute方法不斷的去向線程池當中提交任務，大致流程如下圖所示：
線程池通過execute函數(shù)不斷的往線程池當中的任務隊列加入任務，而線程池當中的線程會不斷的從任務隊列當中取出任務，然后進行執(zhí)行，然后繼續(xù)取任務，繼續(xù)執(zhí)行....，線程的執(zhí)行過程如下：
while (true) { Runnable runnable = taskQueue.take(); // 從任務隊列當中取出任務 runnable.run(); // 執(zhí)行任務}根據(jù)上面所談到的內容，現(xiàn)在我們的需求很清晰了，首先我們需要有一個隊列去存儲我們所需要的任務，然后需要開啟多個線程不斷的去任務隊列當中取出任務，然后進行執(zhí)行，然后重復取任務執(zhí)行任務的操作。
工具介紹
在我們前面提到的線程池實現(xiàn)的原理當中有一個非常重要的數(shù)據(jù)結構，就是ArrayBlockingQueue阻塞隊列，它是一個并發(fā)安全的數(shù)據(jù)結構，我們首先先簡單介紹一下這個數(shù)據(jù)結構的使用方法。（如果你想深入了解阻塞隊列的實現(xiàn)原理，可以參考這篇文章JDK數(shù)組阻塞隊列源碼剖析）
我們主要用的是ArrayBlockingQueue的下面兩個方法：
- put函數(shù)，這個函數(shù)是往線程當中加入數(shù)據(jù)的。我們需要了解的是，如果一個線程調用了這個函數(shù)往隊列當中加入數(shù)據(jù)，如果此時隊列已經滿了則線程需要被掛起，如果沒有滿則需要將數(shù)據(jù)加入到隊列當中，也就是將數(shù)據(jù)存儲到數(shù)組當中。
- take函數(shù)，從隊列當中取出數(shù)據(jù)，但是當隊列為空的時候需要將調用這個方法的線程阻塞。當隊列當中有數(shù)據(jù)的時候，就可以從隊列當中取出數(shù)據(jù)。
- 需要注意的是，如果一個線程被上面兩個任何一個線程阻塞之后，可以調用對應線程的interrupt方法終止線程的執(zhí)行，同時還會拋出一個異常。
下面是一份測試代碼：
import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.TimeUnit; public class QueueTest { public static void main(String[] args) throws InterruptedException { ArrayBlockingQueue queue = new ArrayBlockingQueue(5); // 隊列的容量為5 Thread thread = new Thread(() -> { for (int i = 0; i < 10; i++) { try { queue.put(i); System.out.println("數(shù)據(jù) " + i + "被加入到隊列當中"); } catch (InterruptedException e) { System.out.println("出現(xiàn)了中斷異常"); // 如果出現(xiàn)中斷異常 則退出 線程就不會一直在 put 方法被掛起了 return; }finally { } } }); thread.start(); TimeUnit.SECONDS.sleep(1); thread.interrupt(); }} 上面代碼輸出結果：
數(shù)據(jù) 0被加入到隊列當中數(shù)據(jù) 1被加入到隊列當中數(shù)據(jù) 2被加入到隊列當中數(shù)據(jù) 3被加入到隊列當中數(shù)據(jù) 4被加入到隊列當中出現(xiàn)了中斷異常
上面代碼的執(zhí)行順序是：
線程thread會將0-4這5個數(shù)據(jù)加入到隊列當中，但是在加入第6個數(shù)據(jù)的時候，阻塞隊列已經滿了，因此在加入數(shù)據(jù)的時候線程thread會被阻塞，然后主線程在休息一秒之后中斷了線程thread，然后線程thread發(fā)生了中斷異常，然后被捕獲進入catch代碼塊，然后函數(shù)返回，線程thread就不會一直被阻塞了，這一點在我們后面寫線程池非常重要?。?！
Worker設計
在前文當中我們已經提到了我們的線程需要不斷的去任務隊列里面取出任務然后執(zhí)行，我們設計一個Worker類去做這件事！
- 首先在類當中肯定需要有一個線程池的任務隊列，因為worker需要不斷的從阻塞隊列當中取出任務進行執(zhí)行。
- 我們用一個isStopped變量表示線程是否需要終止了，也就是線程池是否需要關閉，如果線程池需要關閉了，那么線程也應該停止了。
- 我們還需要有一個變量記錄執(zhí)行任務的線程，因為當我們需要關閉線程池的時候需要等待任務隊列當中所有的任務執(zhí)行完成，那么當所有的任務都執(zhí)行執(zhí)行完成的時候，隊列肯定是空的，而如果這個時候有線程還去取任務，那么肯定會被阻塞，前面已經提到了ArrayBlockingQueue的使用方法了，我們可以使用這個線程的interrupt的方法去中斷這個線程的執(zhí)行，這個線程會出現(xiàn)異常，然后這個線程捕獲這個異常就可以退出了，因此我們需要知道對那個線程執(zhí)行interrupt方法！
Worker實現(xiàn)的代碼如下：
import java.util.concurrent.ArrayBlockingQueue; public class Worker implements Runnable { // 用于保存任務的隊列 private ArrayBlockingQueue tasks; // 線程的狀態(tài) 是否終止 private volatile boolean isStopped; // 保存執(zhí)行 run 方法的線程 private volatile Thread thisThread; public Worker(ArrayBlockingQueue tasks) { // 這個參數(shù)是線程池當中傳入的 this.tasks = tasks; } @Override public void run() { thisThread = Thread.currentThread(); while (!isStopped) { try { Runnable task = tasks.take(); task.run(); } catch (InterruptedException e) { // do nothing } } } // 注意是其他線程調用這個方法 同時需要注意是 thisThread 這個線程在執(zhí)行上面的 run 方法 // 其他線程調用 thisThread 的 interrupt 方法之后 thisThread 會出現(xiàn)異常 然后就不會一直阻塞了 // 會判斷 isStopped 是否為 true 如果為 true 的話就可以退出 while 循環(huán)了 public void stop() { isStopped = true; thisThread.interrupt(); // 中斷線程 thisThread } public boolean isStopped(){ return isStopped; }} 線程池設計
- 首先線程池需要可以指定有多少個線程，阻塞隊列的最大長度，因此我們需要有這兩個參數(shù)。
- 線程池肯定需要有一個隊列去存放通過submit函數(shù)提交的任務。
- 需要有一個變量存儲所有的woker，因為線程池關閉的時候需要將這些worker都停下來，也就是調用worker的stop方法。
- 需要有一個shutDown函數(shù)表示關閉線程池。
- 需要有一個函數(shù)能夠停止所有線程的執(zhí)行，因為關閉線程池就是讓所有線程的工作停下來。
線程池實現(xiàn)代碼：
import java.util.ArrayList;import java.util.concurrent.ArrayBlockingQueue; public class MyFixedThreadPool { // 用于存儲任務的阻塞隊列 private ArrayBlockingQueue taskQueue; // 保存線程池當中所有的線程 private ArrayList threadLists; // 線程池是否關閉 private boolean isShutDown; // 線程池當中的線程數(shù)目 private int numThread; public MyFixedThreadPool(int i) { this(Runtime.getRuntime().availableProcessors() + 1, 1024); } public MyFixedThreadPool(int numThread, int maxTaskNumber) { this.numThread = numThread; taskQueue = new ArrayBlockingQueue<>(maxTaskNumber); // 創(chuàng)建阻塞隊列 threadLists = new ArrayList<>(); // 將所有的 worker 都保存下來 for (int i = 0; i < numThread; i++) { Worker worker = new Worker(taskQueue); threadLists.add(worker); } for (int i = 0; i < threadLists.size(); i++) { new Thread(threadLists.get(i), "ThreadPool-Thread-" + i).start(); // 讓worker開始工作 } } // 停止所有的 worker 這個只在線程池要關閉的時候才會調用 private void stopAllThread() { for (Worker worker : threadLists) { worker.stop(); // 調用 worker 的 stop 方法 讓正在執(zhí)行 worker 當中 run 方法的線程停止執(zhí)行 } } public void shutDown() { // 等待任務隊列當中的任務執(zhí)行完成 while (taskQueue.size() != 0) { // 如果隊列當中還有任務 則讓出 CPU 的使用權 Thread.yield(); } // 在所有的任務都被執(zhí)行完成之后 停止所有線程的執(zhí)行 stopAllThread(); } public void submit(Runnable runnable) { try { taskQueue.put(runnable); // 如果任務隊列滿了， 調用這個方法的線程會被阻塞 } catch (InterruptedException e) { e.printStackTrace(); } }} 測試代碼：
public class Test { public static void main(String[] args) { MyFixedThreadPool pool = new MyFixedThreadPool(5, 1024); // 開啟5個線程 任務隊列當中最多只能存1024個任務 for (int i = 0; i < 1000000; i++) { pool.submit(() -> { System.out.println(Thread.currentThread().getName()); // 提交的任務就是打印線程自己的名字 }); } pool.shutDown(); }}上面的代碼是可以正常執(zhí)行并且結束的，這個輸出太長了這里只列出部分輸出結果：
ThreadPool-Thread-0ThreadPool-Thread-4ThreadPool-Thread-0ThreadPool-Thread-1ThreadPool-Thread-3ThreadPool-Thread-1ThreadPool-Thread-3ThreadPool-Thread-3ThreadPool-Thread-3ThreadPool-Thread-3ThreadPool-Thread-3ThreadPool-Thread-2ThreadPool-Thread-3ThreadPool-Thread-2ThreadPool-Thread-1ThreadPool-Thread-0ThreadPool-Thread-0ThreadPool-Thread-0ThreadPool-Thread-1ThreadPool-Thread-4
從上面的輸出我們可以看見線程池當中只有5個線程，這5個線程在不斷從隊列當中取出任務然后執(zhí)行，因為我們可以看到同一個線程的名字輸出了多次。
總結
在本篇文章當中主要介紹了線程池的原理，以及我們應該去如何設計一個線程池，同時也介紹了在阻塞隊列當中一個非常重要的數(shù)據(jù)結構ArrayBlockingQueue的使用方法。
線程池當中有一個阻塞隊列去存放所有被提交到線程池當中的任務。
所有的Worker會不斷的從任務隊列當中取出任務然后執(zhí)行。
線程池的shutDown方法其實比較難思考該怎么實現(xiàn)的，首先在我們真正關閉線程池之前需要將任務隊列當中所有的任務執(zhí)行完成，然后將所有的線程停下來。
在所有的任務執(zhí)行完成之后，可能有的線程就會阻塞在take方法上（從隊列當中取數(shù)據(jù)的方法，如果隊列為空線程會阻塞），好在ArrayBlockingQueue在實現(xiàn)的時候就考慮到了這個問題，只需要其他線程調用了這個被阻塞線程的interrupt方法的話，線程就可以通過捕獲異?；謴蛨?zhí)行，然后判斷isStopped，如果需要停止了就跳出while循環(huán)，這樣的話我們就可以完成所有線程的停止操作了。
名稱欄目：徹底了解線程池的原理—40行從零開始自己寫線程池
網站路徑：http://m.5511xx.com/article/dpjgcdi.html
