Kotlin：該如何實現多線程同步？

問題背景
需執行多線程任務：任務1、任務2並行執行；等全部執行完成后，執行任務3。

// 每個 任務 通過 sleep 模擬耗時 val task1: () -> String = { sleep(2000) "Hello".also { println("task1 finished: $it") } } val task2: () -> String = { sleep(2000) "World".also { println("task2 finished: $it") } } val task3: (String, String) -> String = { p1, p2 -> sleep(2000) "$p1 $p2".also { println("task3 finished: $it") } } 

實現方式
「多線程同步」。Kotlin實現多線程同步的方式主要包括：（含Java實現方式）

方式1：Thread.join
方式2：線程鎖：Synchronized、ReentrantLock、CountDownLatch、CyclicBarrier
方式3：CAS
方式4：Future（CompletableFuture）
方式5：Rxjava
方式6：協程Coroutine、Flow
方式1：Thread.join()
這是最簡單的線程同步方式

@Test
fun test_join() {
    lateinit var s1: String lateinit var s2: String val t1 = Thread { s1 = task1() } val t2 = Thread { s2 = task2() } t1.start() t2.start() t1.join() t2.join() task3(s1, s2) } 

方式2：線程鎖
主要包括：Synchronized、ReentrantLock、CountDownLatch、CyclicBarrier

Synchronized

 @Test fun test_synchrnoized() { lateinit var s1: String lateinit var s2: String Thread { synchronized(Unit) { s1 = task1() } }.start() s2 = task2() synchronized(Unit) { task3(s1, s2) } } 

這里需要特別注意的是：為了同步多個並行任務的結果則需要聲明n個鎖， 即需嵌套n個 synchronized

ReentrantLock
相對於Synchronized，ReentrantLock的使用則不會出現嵌套 synchrnoized 的問題，但仍需創建多個 lock 從而管理多個不同的線程任務。

fun test_ReentrantLock() {

    lateinit var s1: String lateinit var s2: String val lock = ReentrantLock() Thread { lock.lock() s1 = task1() lock.unlock() }.start() s2 = task2() lock.lock() task3(s1, s2) lock.unlock() } 

這里需要額外說明的是，阻塞隊列BlockingQueue內部是通過ReentrantLock實現的，所以其也能實現線程同步，但其應用場景是：生產/消費場景中的同步

fun test_blockingQueue() { lateinit var s1: String lateinit var s2: String val queue = SynchronousQueue<Unit>() Thread { s1 = task1() queue.put(Unit) }.start() s2 = task2() queue.take() task3(s1, s2) } 

CountDownLatch
JUC 中的鎖大都基於 AQS 實現的，可以分為獨享鎖和共享鎖。ReentrantLock 就是一種獨享鎖。相比之下，共享鎖更適合本場景，不需為了每個任務都創建單獨的鎖。

 @Test
    fun test_countdownlatch() {

        lateinit var s1: String lateinit var s2: String val cd = CountDownLatch(2) Thread() { s1 = task1() cd.countDown() }.start() Thread() { s2 = task2() cd.countDown() }.start() cd.await() task3(s1, s2) } 

CyclicBarrier
原理：讓一組線程到達一個同步點后再一起繼續運行，其中任意一個線程未達到同步點，其他已到達的線程均會被阻塞。

 @Test
    fun test_CyclicBarrier() {

        lateinit var s1: String lateinit var s2: String val cb = CyclicBarrier(3) Thread { s1 = task1() cb.await() }.start() Thread() { s2 = task1() cb.await() }.start() cb.await() task3(s1, s2) } 

需要特別注意的是：與 CountDownLatch 的區別在於 CountDownLatch 是一次性的，而 CyclicBarrier 可以被重置后循環利用

方式3：CAS
原理：基於 CAS 的原子類計數
應用場景：一些cpu密集型的短任務同步（因為會比較損耗資源）

fun test_cas() {

    lateinit var s1: String lateinit var s2: String val cas = AtomicInteger(2) Thread { s1 = task1() cas.getAndDecrement() }.start() Thread { s2 = task2() cas.getAndDecrement() }.start() while (cas.get() != 0) {} task3(s1, s2) } 

這里需要特別說明的是，看到 CAS 的無鎖實現，很多人會想到 volatile：並非線程安全，因為volatile 能保證可見性，但是不能保證原子性，cnt-- 並非線程安全，需要加鎖操作

fun test_Volatile() {
    lateinit var s1: String lateinit var s2: String Thread { s1 = task1() cnt-- }.start() Thread { s2 = task2() cnt-- }.start() while (cnt != 0) { } task3(s1, s2) } 

方式4：Future
Java 1.5 開始提供了一種可以在任務執行結束時返回結果的線程同步方式：Callable 和 Future 。即不需通過定義變量來記錄結果了。

// 通過 `future.get()`，可以同步等待結果返回，寫起來非常方便 fun test_future() { val future1 = FutureTask(Callable(task1)) val future2 = FutureTask(Callable(task2)) Executors.newCachedThreadPool().execute(future1) Executors.newCachedThreadPool().execute(future2) task3(future1.get(), future2.get()) } 

這里需要特別說明的是，future.get() 雖然方便，但是會阻塞線程。所以在 Java 8 中引入了 CompletableFuture ：他實現了 Future 接口的同時實現了 CompletionStage 接口，即可針對多個 CompletionStage 進行邏輯組合、實現復雜的異步編程。以回調的形式避免了線程阻塞

fun test_CompletableFuture() { CompletableFuture.supplyAsync(task1) .thenCombine(CompletableFuture.supplyAsync(task2)) { p1, p2 -> task3(p1, p2) }.join() } 

方式5：RxJava
RxJava 提供線程同步操作符：

1.subscribeOn 用來啟動異步任務
2.zip 操作符可以組合兩個 Observable 的結果

fun test_Rxjava() { Observable.zip( Observable.fromCallable(Callable(task1)) .subscribeOn(Schedulers.newThread()), Observable.fromCallable(Callable(task2)) .subscribeOn(Schedulers.newThread()), BiFunction(task3) ).test().awaitTerminalEvent() } 

方式6協程：Coroutine、Flow
Coroutine 是 Kotlin 特有的線程同步方式（前面的方式，其實都是 Java 包本身的線程同步方式。）

fun test_coroutine() {

    runBlocking {
        val c1 = async(Dispatchers.IO) { task1() } val c2 = async(Dispatchers.IO) { task2() } task3(c1.await(), c2.await()) } } 

這里需要特別介紹的是，Kotlin版的 RxJava-協程加強版Flow，使用方式類似RxJava 的操作符，如 zip:

fun test_flow() {

    val flow1 = flow<String> { emit(task1()) } val flow2 = flow<String> { emit(task2()) } runBlocking { flow1.zip(flow2) { t1, t2 -> task3(t1, t2) }.flowOn(Dispatchers.IO) .collect() // flowOn 使得 Task 在異步計算並發射結果。 } }

最后：后續會持續更新哦，喜歡的話就點贊關注一下吧。
相關視頻
[Android進階]kotlin的泛型進階