第8篇筆記在引入join使用場景的時候，有個信息采集功能的案例：有若干臺采集服務器，然后還有一臺主機，這臺主機需要等待這若干臺服務器信息采集完之后再做進一步的處理，一臺采集服務器就對應一個線程，如之前寫的代碼： ```java public class ThreadJoin3 { public static void main(String[] args) throws InterruptedException { long startTimestamp = System.currentTimeMillis(); // 假設有三臺機器，開啟三個線程。 Thread m1 = new Thread(new CaptureRunnable("M1", 1_000L)); Thread m2 = new Thread(new CaptureRunnable("M2", 2_000L)); Thread m3 = new Thread(new CaptureRunnable("M3", 3_000L)); m1.start(); m2.start(); m3.start(); m1.join(); m2.join(); m3.join(); long endTimestamp = System.currentTimeMillis(); System.out.printf("Save data begin timestamp is %s, end timestamp is %s\n", startTimestamp, endTimestamp); System.out.printf("Spend time is %s", endTimestamp - startTimestamp); } } /** * 采集服務器節點的任務。 */ class CaptureRunnable implements Runnable { // 機器節點的名稱 private String machineName; // 采集花費時間 private long spendTime; public CaptureRunnable(String machineName, long spendTime) { this.machineName = machineName; this.spendTime = spendTime; } @Override public void run() { // do the really capture data. try { Thread.sleep(spendTime); System.out.printf(machineName + " completed data capture at timestamp [%s] and successful.\n", System.currentTimeMillis()); } catch (InterruptedException e) { e.printStackTrace(); } } public String getResult() { return machineName + " finish."; } } ``` 為了獲取三個線程統一的結束時間，使用了join，這是三個線程三個服務器的情況，如果有成千上萬臺那怎么辦，不可能每臺服務器對應一個線程，因為線程有一個stackSize的上限，此時就需要用到線程同步來避免這個問題了，我們一步一步完善這個功能： 首先創建十個線程，這里采用流的方式來創建，如下： ```java public class CaptureService { public static void main(String[] args) { Stream.of("M1", "M2", "M3", "M4", "M5", "M6", "M7", "M8", "M9", "M10") .map(CaptureService::createCaptureService) .forEach(t -> { t.start(); try { t.join(); } catch (InterruptedException e) { e.printStackTrace(); } }); Optional.of("All of capture work finished.").ifPresent(System.out::println); } public static Thread createCaptureService(String name) { return new Thread(() -> { //TODO System.out.println(Thread.currentThread().getName()); }, name); } } ``` 運行效果如下：  可以看到線程都跑起來了，但是這樣實際是有問題的，因為寫在foreach里是只等待當前線程執行完了而不是所有的線程執行完。  這里使用java8的Stream.forEach()，因為這個方法是一個teminal operation，這個后續再深入學習，先知道有這么個東西，運行效果如下：  接下來完善createCaptureService： 1. 我們在CaptureService增加一個MAX_WORKER來代表最大執行線程數。 ```java private static final int MAX_WORKER = 5; ``` 2. 定義一個class，作為執行代碼鎖的控制器Control。然后放在一個鏈表里。 ```java private static final LinkedList<Control> CONTROLS = new LinkedList<>(); ``` 3. 如果鏈表的長度大于最大執行線程數，就wait(); ```java while (CONTROLS.size() > MAX_WORKER) { try { CONTROLS.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } ``` 4. 如果鏈表長度小于最大執行線程數，就代表鏈表里還有空間，那么就在鏈表里增加一個Control的對象。 ```java CONTROLS.addLast(new Control()); ``` 5. 然后寫采集的業務，采集業務執行后，先把鏈表里的第一個移除。也就是先進先出。同時notifyAll()wait的線程，這時候一個線程搶導鎖就會繼續執行上面的4步的操作，依次類推下去： ```java synchronized (CONTROLS) { Optional.of("The worker [" + Thread.currentThread().getName() + "] END capture data.").ifPresent(System.out::println); CONTROLS.removeFirst(); CONTROLS.notifyAll(); } ``` 整體代碼如下： ```java /** * @program: ThreadDemo * @description: 數據采集功能：利用多個線程采集多臺服務器運行狀態信息。 * 當服務器數量較少時，可以采取一個線程采集一臺服務器； * 但是服務器數量非常大時，將不可能采取這種方式。 * 可以開啟一定數量的線程采集完成后再采集其他服務器，即運行的線程始終保持著穩定數量。 * @author: hs96.cn@Gmail.com * @create: 2020-09-08 */ public class CaptureService { private static final int MAX_WORKER = 5; private static final LinkedList<Control> CONTROLS = new LinkedList<>(); public static void main(String[] args) { List<Thread> worker = new ArrayList<>(); Stream.of("M1", "M2", "M3", "M4", "M5", "M6", "M7", "M8", "M9", "M10") .map(CaptureService::createCaptureService) .forEach(t -> { t.start(); worker.add(t); }); worker.stream().forEach(t -> { try { t.join(); } catch (InterruptedException e) { e.printStackTrace(); } }); Optional.of("All of capture work finished.").ifPresent(System.out::println); } public static Thread createCaptureService(String name) { return new Thread(() -> { // Optional可以防止NPE空指針異常 Optional.of("The worker [" + Thread.currentThread().getName() + "] BEGIN capture data.").ifPresent(System.out::println); synchronized (CONTROLS) { while (CONTROLS.size() > MAX_WORKER) { try { CONTROLS.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } CONTROLS.addLast(new Control()); } Optional.of("The worker [" + Thread.currentThread().getName() + "] is WORKING...").ifPresent(System.out::println); try { Thread.sleep(10_000); } catch (InterruptedException e) { e.printStackTrace(); } synchronized (CONTROLS) { Optional.of("The worker [" + Thread.currentThread().getName() + "] END capture data.").ifPresent(System.out::println); CONTROLS.removeFirst(); CONTROLS.notifyAll(); } }, name); } private static class Control{ } } ``` 運行結果如下：