探秘多线程-闭锁、栅栏与异步编排

无论是项目开发还是开源代码阅读，多线程都是不可或缺的一个重要知识点，基于这个考量，于是总结出本篇文章，讨论闭锁（CountDownLatch）、栅栏（CyclicBarrier）与异步编排(CompletableFuture)
@Author：Akai-yuan
@更新时间：2023/2/4

1.适用场景

1. 协调子线程结束动作：等待所有子线程运行结束

主线程创建了5个子线程，各子任务执行确认动作，期间主线程进入等待状态，直到各子线程的任务均已经完成，主线程恢复继续执行。

2. 协调子线程开始动作：统一各线程动作开始的时机

从多线程的角度看，这恰似你创建了一些多线程，但是你需要统一管理它们的任务开始时间。

2.设计思想

CountDownLatch基于一个同步器实现，并且只有CountDownLatch(int count)一个构造器，指定数量count不得在中途修改它。
核心函数

• await()：等待latch降为0；
• boolean await(long timeout, TimeUnit unit)：等待latch降为0，但是可以设置超时时间。
• countDown()：latch数量减1；
• getCount()：获取当前的latch数量。

3.场景实例

场景1. 对各子线程的等待

public static void main(String[] args) throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(4);

    Thread t1 = new Thread(countDownLatch::countDown);
    Thread t2 = new Thread(countDownLatch::countDown);
    Thread t3 = new Thread(countDownLatch::countDown);
    Thread t4 = new Thread(() -> {
        try {
            // 稍等...
            Thread.sleep(1500);
            countDownLatch.countDown();
        } catch (InterruptedException ignored) {}
    });

    t1.start();
    t2.start();
    t3.start();
    t4.start();
		//直到所有线程都对计数器进行减一后，这里才放行
    countDownLatch.await();
    System.out.println("所有子线程就位，可以继续执行其他任务");
}

场景2. 对多线程的统一管理

我们仍然用4个线程调用了start()，但是它们在运行时都在等待countDownLatch的信号，在信号未收到前，它们不会往下执行。

public static void main(String[] args) throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(1);

    Thread t1 = new Thread(() -> waitForCountDown(countDownLatch));
    Thread t2 = new Thread(() -> waitForCountDown(countDownLatch));
    Thread t3 = new Thread(() -> waitForCountDown(countDownLatch));
    Thread t4 = new Thread(() -> waitForCountDown(countDownLatch));

    t1.start();
    t2.start();
    t3.start();
    t4.start();
    Thread.sleep(1000);
    countDownLatch.countDown();
    System.out.println("所有线程准备完成");
}

private static void waitForCountDown(CountDownLatch countDownLatch) {
    try {
			// 等待信号
        countDownLatch.await(); 
        System.out.println("本线程等待完毕");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

输出：

所有线程准备完成
本线程等待完毕
本线程等待完毕
本线程等待完毕
本线程等待完毕

Process finished with exit code 0

场景3. SOFAJRaft的实践

// 定义一个CountDownLatch计数器
private final CountDownLatch startTimeInitialized = new CountDownLatch(1);

public void start() {
    switch (workerStateUpdater.get(this)) {
        case WORKER_STATE_INIT:
            if (workerStateUpdater.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
                //此处调用工作线程执行CountDownLatch的countDown()方法
                //即startTimeInitialized.countDown();
                workerThread.start();
            }
            break;
        case WORKER_STATE_STARTED:
            break;
        case WORKER_STATE_SHUTDOWN:
            throw new IllegalStateException("cannot be started once stopped");
        default:
            throw new Error("Invalid WorkerState");
    }

    // 等待startTime被工作线程初始化完成
    while (startTime == 0) {
        try {
            startTimeInitialized.await();
        } catch (InterruptedException ignore) {
            // Ignore - it will be ready very soon.
        }
    }
}

2.CyclicBarrier

1.适用场景

栅栏类似于闭锁，它能阻塞一组线程直到某个事件的发生。栅栏与闭锁的关键区别在于，所有的线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待事件，而栅栏用于等待其他线程。
CyclicBarrier与CountDownLatch的区别

CyclicBarrier	CountDownLatch
CyclicBarrier是可重用的，其中的线程会等待所有的线程完成任务。届时，屏障将被拆除，并可以选择性地做一些特定的动作。	CountDownLatch是一次性的，不同的线程在同一个计数器上工作，直到计数器为0
CyclicBarrier面向的是线程数	CountDownLatch面向的是任务数
在使用CyclicBarrier时，你必须在构造中指定参与协作的线程数，这些线程必须调用await()方法	使用CountDownLatch时，则必须要指定任务数，至于这些任务由哪些线程完成无关紧要
CyclicBarrier可以在所有的线程释放后重新使用	CountDownLatch在计数器为0时不能再使用
在CyclicBarrier中，如果某个线程遇到了中断、超时等问题时，则处于await的线程都会出现问题	在CountDownLatch中，如果某个线程出现问题，其他线程不受影响

2.设计思想

1.构造器

// 指定参与方的数量；
public CyclicBarrier(int parties);
// 指定参与方的数量，并指定在本代次结束时运行的代码
public CyclicBarrier(int parties, Runnable barrierAction)：

2.核心方法

//如果当前线程不是第一个到达屏障的话，它将会进入等待，直到其他线程都到达
//除非发生被中断、屏障被拆除、屏障被重设等情况
public int await();
//和await()类似，但是加上了时间限制；
public int await(long timeout, TimeUnit unit);
//当前屏障是否被拆除；
public boolean isBroken();
//重设当前屏障。会先拆除屏障再设置新的屏障
public void reset();
//正在等待的线程数量
public int getNumberWaiting();

3.场景实例

下面以一个简单的日常对话来讲解CyclicBarrier的使用实例

private static String appointmentPlace = "书房";

public static void main(String[] args) {
  CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> System.out.println("yuan所在的地点:" + appointmentPlace));
    // 线程Akai
    Thread Akai = newThread("Akai", () -> {
    System.out.println("yuan,饭好了快来吃饭...");
    try {
      // 此时Akai在屏障前等待
      cyclicBarrier.await();
      System.out.println("yuan，你来了...");
      // 开始吃饭...
      Thread.sleep(2600); 
      System.out.println("好的，你去洗你的碗吧！");
      // 第二次调用await
      cyclicBarrier.await(); 
      Thread.sleep(100);
      System.out.println("好吧，你这个懒猪！");
    } catch (Exception e) {
      e.printStackTrace();
    }
  });
  // 线程yuan
  Thread yuan = newThread("yuan", () -> {
    try {
      // yuan在敲代码
      Thread.sleep(500); 
      System.out.println("我在敲代码，我马上就来!");
      // yuan到达饭桌前
      cyclicBarrier.await(); 
      Thread.sleep(500); 
      System.out.println("Akai，不好意思，刚刚沉迷于敲代码了！");
      // 开始吃饭...
      Thread.sleep(1500); 
      // yuan想先吃完赶快洗碗然后溜出去敲代码
      System.out.println("我吃完了，我要去洗碗了"); 
      // yuan把地点改成了厨房
      appointmentPlace = "厨房"; 
      // 洗碗中...
      Thread.sleep(1500); 
      System.out.println("︎yuan终于洗完自己的碗了");
      // 第二次调用await
      cyclicBarrier.await();
      System.out.println("Akai你吃完了，你的碗自己去洗吧，我已经在敲代码了");
    } catch (Exception ignored) {}
  });

  Akai.start();
  yuan.start();
}

输出结果：

yuan,饭好了快来吃饭...
我在敲代码，我马上就来!
yuan所在的地点:书房
yuan，你来了...
Akai，不好意思，刚刚沉迷于敲代码了！
我吃完了，我要去洗碗了
好的，你去洗你的碗吧！
yuan终于洗完自己的碗了
yuan所在的地点:厨房
Akai你吃完了，你的碗自己去洗吧，我已经在敲代码了
好吧，你这个懒猪！    

3.CompletableFuture

1.设计思想

1.Future的局限性

• 并发执行多任务：Future只提供了get()方法来获取结果，并且是阻塞的。所以，除了等待你别无他法；
• 无法对多个任务进行链式调用：如果你希望在计算任务完成后执行特定动作，比如发邮件，但Future却没有提供这样的能力；
• 无法组合多个任务：如果你运行了10个任务，并期望在它们全部执行结束后执行特定动作，那么在Future中这是无能为力的；
• 没有异常处理：Future接口中没有关于异常处理的方法；

2.Completable有哪些优势

CompletableFuture是Future接口的扩展和增强。
CompletableFuture完整地继承了Future接口，并在此基础上进行了丰富地扩展，完美地弥补了Future上述的种种问题。更为重要的是，CompletableFuture实现了对任务的编排能力。借助这项能力，我们可以轻松地组织不同任务的运行顺序、规则以及方式。
从某种程度上说，这项能力是它的核心能力。而在以往，虽然通过CountDownLatch等工具类也可以实现任务的编排，但需要复杂的逻辑处理，不仅耗费精力且难以维护。

2.核心设计

我们首先来讨论CompletableFuture的核心：CompletionStage
顾名思义，根据CompletionStage名字中的"Stage"，你可以把它理解为任务编排中的步骤。步骤，即任务编排的基本单元，它可以是一次纯粹的计算或者是一个特定的动作。在一次编排中，会包含多个步骤，这些步骤之间会存在依赖、链式和组合等不同的关系，也存在并行和串行的关系。这种关系，类似于Pipeline或者流式计算。
既然是编排，就需要维护任务的创建、建立计算关系。为此，CompletableFuture提供了多达50多个方法，但没有必要全部完全理解，但我们可以通过分类的方式简化对方法的理解，理解了类型和变种，基本上我们也就掌握了CompletableFuture的核心能力。
这些方法可以总结为以下四类，其他大部分方法都是基于这四种类型的变种:

3.核心用法

1.runAsync

• runAsync()是CompletableFuture最常用的方法之一，它可以接收一个待运行的任务并返回一个CompletableFuture
• 当我们想异步运行某个任务时，在以往需要手动实现Thread或者借助Executor实现。而通过runAsync()`就简单多了。比如，我们可以直接传入Runnable类型的任务：

CompletableFuture.runAsync(new Runnable() {
    @Override
    public void run() {
        System.out.println("something");
    }
});

2.supply与supplyAsync

• 所谓supply表示提供结果，换句话说当我们使用supply()时，就表明我们会返回一个结果，并且这个结果可以被后续的任务所使用。

// 创建nameFuture，返回姓名
CompletableFuture <String> nameFuture = CompletableFuture.supplyAsync(() -> {
    return "Akai-yuan";
});

 // 使用thenApply()接收nameFuture的结果，并执行回调动作
CompletableFuture <String> sayLoveFuture = nameFuture.thenApply(name -> {
    return "love you，" + name;
});

//阻塞获得表白的结果
System.out.println(sayLoveFuture.get()); // love you，Akai-yuan

一旦理解了supply()的含义，它也就如此简单。如果你希望用新的线程运行任务，可以使用supplyAsync().

3.thenApply与thenApplyAsync

• 我们已经知道supply()是用于提供结果的，并且顺带提了thenApply()。很明显，thenApply()是supply()的搭档，用于接收supply()的执行结果，并执行特定的代码逻辑，最后返回CompletableFuture结果。

 // 使用thenApply()接收nameFuture的结果，并执行回调动作
CompletableFuture <String> sayLoveFuture = nameFuture.thenApply(name -> {
    return "爱你，" + name;
});

public <U> CompletableFuture <U> thenApplyAsync(
    Function <? super T, ? extends U> fn) {
    return uniApplyStage(null, fn);
}

4.thenAccept与thenAcceptAsync

作为supply()的档案，thenApply()并不是唯一的存在，thenAccept()也是。但与thenApply()不同，thenAccept()只接收数据，但不会返回，它的返回类型是Void.

CompletableFuture<Void> sayLoveFuture = nameFuture.thenAccept(name -> {
     System.out.println("爱你，" + name);
});
        
public CompletableFuture < Void > thenAccept(Consumer < ? super T > action) {
    return uniAcceptStage(null, action);
}

5.thenRun

thenRun()就比较简单了，不接收任务的结果，只运行特定的任务，并且也不返回结果。

public CompletableFuture < Void > thenRun(Runnable action) {
   return uniRunStage(null, action);
}

所以，如果你在回调中不想返回任何的结果，只运行特定的逻辑，那么你可以考虑使用thenAccept和thenRun一般来说，这两个方法会在调用链的最后面使用。

6.thenCompose与 thenCombine

以上几种方法都是各玩各的，但thenCompose()与thenCombine()就不同了，它们可以实现对依赖和非依赖两种类型的任务的编排。
编排两个存在依赖关系的任务
在前面的例子中，在接收前面任务的结果时，我们使用的是thenApply(). 也就是说，sayLoveFuture在执行时必须依赖nameFuture的完成，否则执行个锤子。

// 创建Future
CompletableFuture <String> nameFuture = CompletableFuture.supplyAsync(() -> {
    return "Akai-yuan";
});

 // 使用thenApply()接收nameFuture的结果，并执行回调动作
CompletableFuture <String> sayLoveFuture = nameFuture.thenApply(name -> {
    return "爱你，" + name;
});

但其实，除了thenApply()之外，我们还可以使用thenCompose()来编排两个存在依赖关系的任务。比如，上面的示例代码可以写成：

// 创建Future
CompletableFuture <String> nameFuture = CompletableFuture.supplyAsync(() -> {
    return "Akai-yuan";
});

CompletableFuture<String> sayLoveFuture2 = nameFuture.thenCompose(name -> {
    return CompletableFuture.supplyAsync(() -> "爱你，" + name);
});

可以看到，thenCompose()和thenApply()的核心不同之处在于它们的返回值类型：

• thenApply()：返回计算结果的原始类型，比如返回String;
• thenCompose()：返回CompletableFuture类型，比如返回CompletableFuture.

组合两个相互独立的任务
考虑一个场景，当我们在执行某个任务时，需要其他任务就绪才可以，应该怎么做？这样的场景并不少见，我们可以使用前面学过的并发工具类实现，也可以使用thenCombine()实现。
举个例子，当我们计算某个胜率时，我们需要获取她参与的总场次（rounds），以及获胜的场次（winRounds），然后再通过winRounds / rounds来计算。对于这个计算，我们可以这么做：

CompletableFuture < Integer > roundsFuture = CompletableFuture.supplyAsync(() -> 500);
CompletableFuture < Integer > winRoundsFuture = CompletableFuture.supplyAsync(() -> 365);

CompletableFuture < Object > winRateFuture = roundsFuture
    .thenCombine(winRoundsFuture, (rounds, winRounds) -> {
        if (rounds == 0) {
            return 0.0;
        }
        DecimalFormat df = new DecimalFormat("0.00");
        return df.format((float) winRounds / rounds);
    });
System.out.println(winRateFuture.get());

thenCombine()将另外两个任务的结果同时作为参数，参与到自己的计算逻辑中。在另外两个参数未就绪时，它将会处于等待状态。

7.allOf与anyOf

allOf()与anyOf()也是一对孪生兄弟，当我们需要对多个Future的运行进行组织时，就可以考虑使用它们：

• allOf()：给定一组任务，等待所有任务执行结束；
• anyOf()：给定一组任务，等待其中任一任务执行结束。

allOf()与anyOf()的方法签名如下：

static CompletableFuture<Void>	 allOf(CompletableFuture<?>... cfs)
static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

需要注意的是，anyOf()将返回完任务的执行结果，但是allOf()不会返回任何结果，它的返回值是Void.
allOf()与anyOf()的示例代码如下所示。我们创建了roundsFuture和winRoundsFuture，并通过sleep模拟它们的执行时间。在执行时，winRoundsFuture将会先返回结果，所以当我们调用 CompletableFuture.anyOf时也会发现输出的是365.

 CompletableFuture < Integer > roundsFuture = CompletableFuture.supplyAsync(() -> {
   try {
     Thread.sleep(200);
     return 500;
   } catch (InterruptedException e) {
     return null;
   }
 });
 CompletableFuture < Integer > winRoundsFuture = CompletableFuture.supplyAsync(() -> {
   try {
     Thread.sleep(100);
     return 365;
   } catch (InterruptedException e) {
     return null;
   }
 });

 CompletableFuture < Object > completedFuture = CompletableFuture.anyOf(winRoundsFuture, roundsFuture);
 System.out.println(completedFuture.get()); // 返回365

 CompletableFuture < Void > completedFutures = CompletableFuture.allOf(winRoundsFuture, roundsFuture);

在CompletableFuture之前，如果要实现所有任务结束后执行特定的动作，我们可以考虑CountDownLatch等工具类。现在，则多了一选项，我们也可以考虑使用CompletableFuture.allOf.

8.异常处理

在CompletableFuture链式调用中，如果某个任务发生了异常，那么后续的任务将都不会再执行。对于异常，我们有两种处理方式：exceptionally()和handle().

1.使用exceptionally()回调处理异常

在链式调用的尾部使用exceptionally()，捕获异常并返回错误情况下的默认值。需要注意的是，exceptionally()仅在发生异常时才会调用。

CompletableFuture < ? extends Serializable > winRateFuture = roundsFuture
    .thenCombine(winRoundsFuture, (rounds, winRounds) -> {
        if (rounds == 0) {
            throw new RuntimeException("总场次错误");
        }
        DecimalFormat df = new DecimalFormat("0.00");
        return df.format((float) winRounds / rounds);
    }).exceptionally(ex -> {
        System.out.println("出错：" + ex.getMessage());
        return "";
    });
System.out.println(winRateFuture.get());

2. 使用handle()处理异常

除了exceptionally()，CompletableFuture也提供了handle()来处理异常。不过，与exceptionally()不同的是，当我们在调用链中使用了handle()，那么无论是否发生异常，都会调用它。所以，在handle()方法的内部，我们需要通过 if (ex != null) 来判断是否发生了异常。

CompletableFuture < ? extends Serializable > winRateFuture = roundsFuture
    .thenCombine(winRoundsFuture, (rounds, winRounds) -> {
        if (rounds == 0) {
            throw new RuntimeException("总场次错误");
        }
        DecimalFormat df = new DecimalFormat("0.00");
        return df.format((float) winRounds / rounds);
    }).handle((res, ex) -> {
        if (ex != null) {
            System.out.println("出错：" + ex.getMessage());
            return "";
        }
        return res;
    });
System.out.println(winRateFuture.get());

当然，如果我们允许某个任务发生异常而不中断整个调用链路，那么可以在其内部通过try-catch消化掉。