我自己面試時被問過兩次多任務(wù)并行相關(guān)的問題：

假設(shè)現(xiàn)在有10個任務(wù)，要求同時處理，并且必須所有任務(wù)全部完成才返回結(jié)果

這個面試題的難點是：

• 既然要同時處理，那么肯定要用多線程。怎么設(shè)計多線程同時處理任務(wù)呢？
• 要求返回結(jié)果，那么就不能用簡單的Thread+Runnable了，這個是無返回結(jié)果的
• 最難的是，這些任務(wù)彼此間還有關(guān)系：任務(wù)全部結(jié)束才算完成

下面3個Demo，CountDownLatch的結(jié)果處理交給大家自行完成。

FutureTask

/*** @author mx*/
public class FutureTaskDemo {private static AtomicInteger count = new AtomicInteger(10);public static void main(String[] args) throws ExecutionException, InterruptedException {// 準(zhǔn)備10個線程ExecutorService executorService = Executors.newFixedThreadPool(10);// 收集每個任務(wù)的結(jié)果List<Future<Integer>> resultList = new ArrayList<>();long start = System.currentTimeMillis();// 并行處理10個任務(wù)for (int i = 0; i < 10; i++) {// 準(zhǔn)備任務(wù)Callable<Integer> task = () -> {// 模擬任務(wù)耗時 0~4秒int seconds = ThreadLocalRandom.current().nextInt(5);TimeUnit.SECONDS.sleep(seconds);System.out.println("task is completed! cost:" + seconds + "s left: " + count.decrementAndGet());// 模擬返回結(jié)果return 1;};// 提交任務(wù)Future<Integer> partResult = executorService.submit(task);// 收集結(jié)果resultList.add(partResult);}int result = 0;// 阻塞獲取并累加結(jié)果for (Future<Integer> future : resultList) {result += future.get();}// 最終全部任務(wù)完成，總耗時取決于最耗時的任務(wù)時長System.out.println("all task is completed! result=" + result + " cost: " + (System.currentTimeMillis() - start) + "ms");}
}

結(jié)果展示

task is completed! cost:0s left: 9

task is completed! cost:1s left: 8

task is completed! cost:1s left: 7

task is completed! cost:2s left: 6

task is completed! cost:3s left: 4

task is completed! cost:3s left: 5

task is completed! cost:3s left: 3

task is completed! cost:3s left: 1

task is completed! cost:3s left: 2

task is completed! cost:4s left: 0

all task is completed! result=10 cost: 4110ms

我原先還寫過另一個復(fù)雜版本：

/*** @author mx*/
public class FutureTaskDemo {private static AtomicInteger count = new AtomicInteger(10);public static void main(String[] args) throws ExecutionException, InterruptedException {// 準(zhǔn)備10個線程ExecutorService executorService = Executors.newFixedThreadPool(10);// 收集任務(wù)List<Callable<Integer>> taskList = new ArrayList<>();// 收集結(jié)果List<Future<Integer>> resultList = new ArrayList<>();long start = System.currentTimeMillis();// 先準(zhǔn)備10個任務(wù)for (int i = 0; i < 10; i++) {Callable<Integer> task = () -> {// 模擬任務(wù)耗時 0~4秒int seconds = ThreadLocalRandom.current().nextInt(5);TimeUnit.SECONDS.sleep(seconds);System.out.println("task is completed! cost:" + seconds + "s left: " + count.decrementAndGet());// 模擬返回結(jié)果return 1;};// 收集任務(wù)taskList.add(task);}// 10個任務(wù)并行執(zhí)行for (int i = 0; i < 10; i++) {// 從任務(wù)列表取出任務(wù)，丟到線程池執(zhí)行Future<Integer> partResult = executorService.submit(taskList.get(i));// 收集異步結(jié)果resultList.add(partResult);}// 最終結(jié)果，用于累加int result = 0;// 是否全部結(jié)束，否則一直循環(huán)等待while (notFinished(resultList)) {// wait for all task to be completed...}// 主線程執(zhí)行到這，肯定全部任務(wù)已經(jīng)結(jié)束，所以get()會立即返回結(jié)果，不會再阻塞等待for (Future<Integer> future : resultList) {result += future.get();}// 最終全部任務(wù)完成，總耗時取決于最耗時的任務(wù)時長System.out.println("all task is completed! result=" + result + " cost: " + (System.currentTimeMillis() - start) + "ms");}/*** 是否全部完成** @param list* @return*/private static boolean notFinished(List<Future<Integer>> list) {for (Future<Integer> future : list) {if (!future.isDone()) {return true;}}return false;}
}

結(jié)果展示

task is completed! cost:0s left: 9

task is completed! cost:0s left: 8

task is completed! cost:2s left: 7

task is completed! cost:3s left: 6

task is completed! cost:3s left: 3

task is completed! cost:3s left: 4

task is completed! cost:3s left: 5

task is completed! cost:4s left: 2

task is completed! cost:4s left: 1

task is completed! cost:4s left: 0

all task is completed! result=10 cost: 4051ms

在當(dāng)前場景下，其實沒必要。

有些人可能覺得第一個版本會出現(xiàn)以下問題：

假設(shè)總共就2個任務(wù)，但是第一個任務(wù)耗時3秒，第二個任務(wù)耗時1秒。第二個任務(wù)的1秒是建立在第一個任務(wù)的3秒后，所以總耗時就變成了4秒。

實際上并不會。當(dāng)future1#get()阻塞獲取第一個任務(wù)結(jié)果的過程中，第二個任務(wù)已經(jīng)完成，所以future2.get()不會再阻塞，而是直接返回結(jié)果。

CountDownLatch

CountDownLatch是什么？學(xué)名叫閉鎖，俗名叫門栓。

/*** @author mx*/
public class CountDownLatchDemo {public static void main(String[] args) throws InterruptedException {// 1.先看簡單demo，了解下CountDownLatchmainThreadAndAsyncThread();// 2.嘗試使用CountDownLatch解決任務(wù)并行問題（不處理結(jié)果）
//        multiThreadTask();}/*** CountDownLatch簡單demo** @throws InterruptedException*/private static void mainThreadAndAsyncThread() throws InterruptedException {// 準(zhǔn)備一個countDownLatch，設(shè)置10，相當(dāng)于給門加了10把鎖CountDownLatch countDownLatch = new CountDownLatch(10);long start = System.currentTimeMillis();// 副線程去處理任務(wù)，每個任務(wù)耗時1秒，每處理完1個任務(wù)就解開一把鎖new Thread(() -> {for (int i = 0; i < 10; i++) {try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}countDownAndPrint(countDownLatch, 1);}}).start();// 一夫當(dāng)關(guān)，萬夫莫開。只要門上10把鎖沒有全部解開，任何線程都別想想往下走countDownLatch.await();System.out.println("all task is completed! cost: " + (System.currentTimeMillis() - start) + "ms");}/*** CountDownLatch應(yīng)用：演示10個任務(wù)并行執(zhí)行，全部完成后返回結(jié)果** @throws InterruptedException*/private static void multiThreadTask() throws InterruptedException {// 準(zhǔn)備一個countDownLatch，設(shè)置10，相當(dāng)于給門加了10把鎖CountDownLatch countDownLatch = new CountDownLatch(10);long start = System.currentTimeMillis();// 啟動10個線程執(zhí)行任務(wù)for (int i = 0; i < 10; i++) {new Thread(() -> {try {// 線程進(jìn)來執(zhí)行任務(wù)，隨機(jī)睡0~4秒int seconds = ThreadLocalRandom.current().nextInt(5);TimeUnit.SECONDS.sleep(seconds);// 每完成一個任務(wù)，解開一把鎖countDownAndPrint(countDownLatch, seconds);} catch (InterruptedException e) {e.printStackTrace();}}).start();}// 一夫當(dāng)關(guān)，萬夫莫開。只要門上10把鎖沒有全部解開，任何線程都別想想往下走countDownLatch.await();System.out.println("all task is completed! cost: " + (System.currentTimeMillis() - start) + "ms");}/*** countDown()并且打印，其實是不需要加synchronized的，這里只是為了多線程環(huán)境下正確打印** @param countDownLatch* @param seconds*/private static synchronized void countDownAndPrint(CountDownLatch countDownLatch, int seconds) {countDownLatch.countDown();System.out.println("task completed, cost: " + seconds + "s left: " + countDownLatch.getCount());}
}

結(jié)果展示

task is completed! cost:0s left: 9

task is completed! cost:0s left: 8

task is completed! cost:0s left: 7

task is completed! cost:2s left: 6

task is completed! cost:2s left: 5

task is completed! cost:2s left: 3

task is completed! cost:2s left: 4

task is completed! cost:3s left: 2

task is completed! cost:4s left: 1

task is completed! cost:4s left: 0

all task is completed! result=10 cost: 4049ms

CompletableFuture

順便說一句，上面的兩個Demo都是鬧著玩呢，實際開發(fā)別傻不拉幾地自己寫哈...會被打，而且是背身單打顏扣的那種。

/*** @author mx*/
public class CompletableFutureDemo {private static AtomicInteger count = new AtomicInteger(10);public static void main(String[] args) throws InterruptedException, ExecutionException {// 準(zhǔn)備10個線程ExecutorService executorService = Executors.newFixedThreadPool(10);// 收集結(jié)果List<CompletableFuture<Integer>> resultList = new ArrayList<>();long start = System.currentTimeMillis();// 任務(wù)并行for (int i = 0; i < 10; i++) {CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {// 模擬任務(wù)耗時 0~4秒try {int seconds = ThreadLocalRandom.current().nextInt(5);TimeUnit.SECONDS.sleep(seconds);System.out.println("task is completed! cost:" + seconds + "s left: " + count.decrementAndGet());} catch (InterruptedException e) {e.printStackTrace();}// 模擬返回結(jié)果return 1;}, executorService);resultList.add(completableFuture);}// 處理結(jié)果int result = 0;for (CompletableFuture<Integer> completableFuture : resultList) {result += completableFuture.get();}// 最終全部任務(wù)完成，總耗時取決于最耗時的任務(wù)時長System.out.println("all task is completed! result=" + result + " cost: " + (System.currentTimeMillis() - start) + "ms");}
}

結(jié)果展示

task is completed! cost:0s left: 9

task is completed! cost:0s left: 8

task is completed! cost:0s left: 7

task is completed! cost:0s left: 6

task is completed! cost:2s left: 5

task is completed! cost:3s left: 4

task is completed! cost:3s left: 3

task is completed! cost:3s left: 2

task is completed! cost:4s left: 1

task is completed! cost:4s left: 0

all task is completed! result=10 cost: 4051ms

實際開發(fā)案例

public class CompletableFutureDemo {private static AtomicInteger count = new AtomicInteger(2);public static void main(String[] args) throws InterruptedException, ExecutionException {// 準(zhǔn)備10個線程ExecutorService executorService = Executors.newFixedThreadPool(2);long start = System.currentTimeMillis();// 模擬處理訂單CompletableFuture<Void> dealOrder = CompletableFuture.runAsync(() -> {// 模擬任務(wù)耗時 0~4秒try {int seconds = ThreadLocalRandom.current().nextInt(5);TimeUnit.SECONDS.sleep(seconds);System.out.println("task is completed! cost:" + seconds + "s left: " + count.decrementAndGet());} catch (InterruptedException e) {e.printStackTrace();}}, executorService);// 模擬處理庫存CompletableFuture<Void> dealStock = CompletableFuture.runAsync(() -> {// 模擬任務(wù)耗時 0~4秒try {int seconds = ThreadLocalRandom.current().nextInt(5);TimeUnit.SECONDS.sleep(seconds);System.out.println("task is completed! cost:" + seconds + "s left: " + count.decrementAndGet());} catch (InterruptedException e) {e.printStackTrace();}}, executorService);// 可變參數(shù)，可以傳任意個CompletableFuture，阻塞等待所有任務(wù)完成CompletableFuture.allOf(dealOrder, dealStock).get();// 最終全部任務(wù)完成，總耗時取決于最耗時的任務(wù)時長System.out.println("all task is completed! cost: " + (System.currentTimeMillis() - start) + "ms");}
}

結(jié)果展示

task is completed! cost:2s left: 1

task is completed! cost:3s left: 0

all task is completed! cost: 3058ms