问题背景
需执行多线程任务:任务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()
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 { emit(task1()) }
val flow2 = flow { emit(task2()) }
runBlocking {
flow1.zip(flow2) { t1, t2 ->
task3(t1, t2)
}.flowOn(Dispatchers.IO)
.collect()
// flowOn 使得 Task 在异步计算并发射结果。
}
}
1、本站所有资源均从互联网上收集整理而来,仅供学习交流之用,因此不包含技术服务请大家谅解!
2、本站不提供任何实质性的付费和支付资源,所有需要积分下载的资源均为网站运营赞助费用或者线下劳务费用!
3、本站所有资源仅用于学习及研究使用,您必须在下载后的24小时内删除所下载资源,切勿用于商业用途,否则由此引发的法律纠纷及连带责任本站和发布者概不承担!
4、本站站内提供的所有可下载资源,本站保证未做任何负面改动(不包含修复bug和完善功能等正面优化或二次开发),但本站不保证资源的准确性、安全性和完整性,用户下载后自行斟酌,我们以交流学习为目的,并不是所有的源码都100%无错或无bug!如有链接无法下载、失效或广告,请联系客服处理!
5、本站资源除标明原创外均来自网络整理,版权归原作者或本站特约原创作者所有,如侵犯到您的合法权益,请立即告知本站,本站将及时予与删除并致以最深的歉意!
6、如果您也有好的资源或教程,您可以投稿发布,成功分享后有站币奖励和额外收入!
7、如果您喜欢该资源,请支持官方正版资源,以得到更好的正版服务!
8、请您认真阅读上述内容,注册本站用户或下载本站资源即您同意上述内容!
原文链接:https://www.dandroid.cn/archives/21324,转载请注明出处。
评论0