Kotlin線程同步的幾種實(shí)現(xiàn)方法

更新時(shí)間：2021年07月09日 08:58:49 作者：fundroid

面試的時(shí)候經(jīng)常會(huì)被問(wèn)及多線程同步的問(wèn)題，在 Kotlin 中我們有多種實(shí)現(xiàn)方式，本文將所有這些方式做了整理，感興趣的小伙伴們可以參考一下

面試的時(shí)候經(jīng)常會(huì)被問(wèn)及多線程同步的問(wèn)題，例如：
“ 現(xiàn)有 Task1、Task2 等多個(gè)并行任務(wù)，如何等待全部執(zhí)行完成后，執(zhí)行 Task3?！?br /> 在 Kotlin 中我們有多種實(shí)現(xiàn)方式，本文將所有這些方式做了整理，建議收藏。
1. Thread.join
2. Synchronized
3. ReentrantLock
4. BlockingQueue
5. CountDownLatch
6. CyclicBarrier
7. CAS
8. Future
9. CompletableFuture
10. Rxjava
11. Coroutine
12. Flow

我們先定義三個(gè)Task，模擬上述場(chǎng)景， Task3 基于 Task1、Task2 返回的結(jié)果拼接字符串，每個(gè) Task 通過(guò) sleep 模擬耗時(shí)：

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") }
}

1. Thread.join()

Kotlin 兼容 Java，Java 的所有線程工具默認(rèn)都可以使用。其中最簡(jiǎn)單的線程同步方式就是使用 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

使用 synchronized 鎖進(jìn)行同步

 @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)
        }
    }

但是如果超過(guò)三個(gè)任務(wù)，使用 synchrnoized 這種寫法就比較別扭了，為了同步多個(gè)并行任務(wù)的結(jié)果需要聲明n個(gè)鎖，并嵌套n個(gè) synchronized。

3. ReentrantLock

ReentrantLock 是 JUC 提供的線程鎖，可以替換 synchronized 的使用

 @Test
    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()
    }

ReentrantLock 的好處是，當(dāng)有多個(gè)并行任務(wù)時(shí)是不會(huì)出現(xiàn)嵌套 synchrnoized 的問(wèn)題，但仍然需要?jiǎng)?chuàng)建多個(gè) lock 管理不同的任務(wù)，

4. BlockingQueue

阻塞隊(duì)列內(nèi)部也是通過(guò) Lock 實(shí)現(xiàn)的，所以也可以達(dá)到同步鎖的效果

 @Test
    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)
    }

當(dāng)然，阻塞隊(duì)列更多是使用在生產(chǎn)/消費(fèi)場(chǎng)景中的同步。

5. CountDownLatch

JUC 中的鎖大都基于 AQS 實(shí)現(xiàn)的，可以分為獨(dú)享鎖和共享鎖。ReentrantLock 就是一種獨(dú)享鎖。相比之下，共享鎖更適合本場(chǎng)景。例如 CountDownLatch，它可以讓一個(gè)線程一直處于阻塞狀態(tài)，直到其他線程的執(zhí)行全部完成：

 @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)
    }

共享鎖的好處是不必為了每個(gè)任務(wù)都創(chuàng)建單獨(dú)的鎖，即使再多并行任務(wù)寫起來(lái)也很輕松

6. CyclicBarrier

CyclicBarrier 是 JUC 提供的另一種共享鎖機(jī)制，它可以讓一組線程到達(dá)一個(gè)同步點(diǎn)后再一起繼續(xù)運(yùn)行，其中任意一個(gè)線程未達(dá)到同步點(diǎn)，其他已到達(dá)的線程均會(huì)被阻塞。
與 CountDownLatch 的區(qū)別在于 CountDownLatch 是一次性的，而 CyclicBarrier 可以被重置后重復(fù)使用，這也正是 Cyclic 的命名由來(lái)，可以循環(huán)使用

 @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)
    }

7. CAS

AQS 內(nèi)部通過(guò)自旋鎖實(shí)現(xiàn)同步，自旋鎖的本質(zhì)是利用 CompareAndSwap 避免線程阻塞的開(kāi)銷。
因此，我們可以使用基于 CAS 的原子類計(jì)數(shù)，達(dá)到實(shí)現(xiàn)無(wú)鎖操作的目的。

  @Test
    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)
    }

while 循環(huán)空轉(zhuǎn)看起來(lái)有些浪費(fèi)資源，但是自旋鎖的本質(zhì)就是這樣，所以 CAS 僅僅適用于一些cpu密集型的短任務(wù)同步。

volatile

看到 CAS 的無(wú)鎖實(shí)現(xiàn)，也許很多人會(huì)想到 volatile，是否也能實(shí)現(xiàn)無(wú)鎖的線程安全？

  @Test
    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)
    }

注意，這種寫法是錯(cuò)誤的
volatile 能保證可見(jiàn)性，但是不能保證原子性，cnt-- 并非線程安全，需要加鎖操作

8. Future

上面無(wú)論有鎖操作還是無(wú)鎖操作，都需要定義兩個(gè)變量s1、s2記錄結(jié)果非常不方便。
Java 1.5 開(kāi)始，提供了 Callable 和 Future ，可以在任務(wù)執(zhí)行結(jié)束時(shí)返回結(jié)果。

@Test
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())

}

通過(guò) future.get()，可以同步等待結(jié)果返回，寫起來(lái)非常方便

9. CompletableFuture

future.get() 雖然方便，但是會(huì)阻塞線程。 Java 8 中引入了 CompletableFuture ，他實(shí)現(xiàn)了 Future 接口的同時(shí)實(shí)現(xiàn)了 CompletionStage 接口。 CompletableFuture 可以針對(duì)多個(gè) CompletionStage 進(jìn)行邏輯組合、實(shí)現(xiàn)復(fù)雜的異步編程。這些邏輯組合的方法以回調(diào)的形式避免了線程阻塞：

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

10. RxJava

RxJava 提供的各種操作符以及線程切換能力同樣可以幫助我們實(shí)現(xiàn)需求：
zip 操作符可以組合兩個(gè) Observable 的結(jié)果；subscribeOn 用來(lái)啟動(dòng)異步任務(wù)

@Test
fun test_Rxjava() {

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

}

11. Coroutine

前面講了那么多，其實(shí)都是 Java 的工具。 Coroutine 終于算得上是 Kotlin 特有的工具了：

@Test
fun test_coroutine() {

    runBlocking {
        val c1 = async(Dispatchers.IO) {
            task1()
        }

        val c2 = async(Dispatchers.IO) {
            task2()
        }

        task3(c1.await(), c2.await())
    }
}

寫起來(lái)特別舒服，可以說(shuō)是集前面各類工具的優(yōu)點(diǎn)于一身。

12. Flow

Flow 就是 Coroutine 版的 RxJava，具備很多 RxJava 的操作符，例如 zip:

@Test
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 在異步計(jì)算并發(fā)射結(jié)果。