关于android: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 的泛型进阶