引自:建议收藏!Kotlin 线程同步的 N 种方法
线程同步的方式
问题: 现有 Task1、Task2 等多个并行任务,如何等待全部任务执行完成后,开始执行 Task3 ?
1 2 3 4 5 6 7 8 9 10 11 12 13 14
| 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") } }
|
Thread.join
join
方法:让当前所在线程挂起,让指定线程先执行
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
| @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) }
|
Synchronized
synchronized
:锁住一个对象,使其他线程无法访问该对象的同步代码块,但可以访问该对象的非同步代码块。
- 修饰代码块:
synchronized(this){}
,锁住当前对象,作用范围是{}代码块
- 修饰方法:锁住当前对象,作用范围是整个方法
- 修饰静态方法:锁住当前类的所有对象,作用范围是整个静态方法
- 修饰类:
synchronized(XXX.class){}
,锁住类的所有对象,作用范围是{}代码块
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
| @Test fun test_synchrnoized() { lateinit var s1: String lateinit var s2: String Thread { synchronized(Unit) { s1 =l task1() } }.start() s2 = task2() synchronized(Unit) { task3(s1, s2) } }
|
多个并行任务写法比较复杂,需要嵌套synchronized
ReentrantLock
ReentrantLock
是JUC(java.util .concurrent,并发工具包)
提供的线程锁,可以替换Synchronized使用
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
| @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() }
|
多个并行任务需要创建多个Lock管理不同的任务
BlockingQueue
阻塞队列:阻塞队列内部也是通过 ReentrantLock 实现的。
阻塞队列更多是使用在生产/消费场景中的同步。
1 2 3 4 5 6 7 8 9 10 11 12 13
| @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) }
|
CountDownLatch
JUC 中的锁大都基于 AQS
实现的,可以分为独享锁和共享锁。ReentrantLock
就是一种独享锁。相比之下,共享锁更适合本场景。例如 CountDownLatch
,它可以让一个线程一直处于阻塞状态,直到其他线程的执行全部完成:
共享锁的好处是不必为了每个任务都创建单独的锁,即使再多并行任务写起来也很轻松
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
| @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
CyclicBarrier
是 JUC 提供的另一种共享锁机制,它可以让一组线程到达一个同步点后再一起继续运行,其中任意一个线程未达到同步点,其他已到达的线程均会被阻塞。
与 CountDownLatch
的区别在于 CountDownLatch
是一次性的,而 CyclicBarrier
可以被重置后重复使用,这也正是 Cyclic
的命名由来,可以循环使用
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
| @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) }
|
CAS
AQS 内部通过自旋锁实现同步,自旋锁的本质是利用 CompareAndSwap
避免线程阻塞的开销。因此,我们可以使用基于 CAS 的原子类计数,达到实现无锁操作的目的。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
| @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
循环空转看起来有些浪费资源,但是自旋锁的本质就是这样,所以 CAS 仅仅适用于一些cpu密集型的短任务同步。
Volatile
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
| @Volatile var cnt = 2 @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) }
|
注意,这种写法是错误的,volatile
能保证可见性,但是不能保证原子性,cnt--
并非线程安全,需要加锁操作
- 获取cnt变量为2
- 操作数-1
- 写回内存
volatile只能保证读取的时候是最新的,当两个线程都读完之后,分别执行了2步骤,再写回内存,结果被覆盖
Future
上面无论有锁操作还是无锁操作,都需要定义两个变量s1
、s2
记录结果非常不方便。Java 1.5 开始,提供了 Callable
和 Future
,可以在任务执行结束时返回结果。
通过 future.get()
,可以同步等待结果返回,写起来非常方便
1 2 3 4 5 6 7 8
| @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()) }
|
CompletableFuture
future.get()
虽然方便,但是会阻塞线程。Java 8 中引入了 CompletableFuture
,他实现了 Future 接口的同时实现了 CompletionStage
接口。CompletableFuture
可以针对多个 CompletionStage
进行逻辑组合、实现复杂的异步编程。这些逻辑组合的方法以回调的形式避免了线程阻塞:
1 2 3 4 5 6 7
| @Test fun test_CompletableFuture() { CompletableFuture.supplyAsync(task1) .thenCombine(CompletableFuture.supplyAsync(task2)) { p1, p2 -> task3(p1, p2) }.join() }
|
RxJava
RxJava
提供的各种操作符以及线程切换能力同样可以帮助我们实现需求:zip
操作符可以组合两个 Observable
的结果;subscribeOn
用来启动异步任务
1 2 3 4 5 6 7 8 9 10
| @Test fun test_Rxjava() { Observable.zip( Observable.fromCallable(Callable(task1)) .subscribeOn(Schedulers.newThread()), Observable.fromCallable(Callable(task2)) .subscribeOn(Schedulers.newThread()), BiFunction(task3) ).test().awaitTerminalEvent() }
|
Coroutine
协程:kotlin
特有
1 2 3 4 5 6 7 8 9 10 11 12
| @Test fun test_coroutine() { runBlocking { val c1 = async(Dispatchers.IO) { task1() } val c2 = async(Dispatchers.IO) { task2() } task3(c1.await(), c2.await()) } }
|
Flow
1 2 3 4 5 6 7 8 9 10 11
| @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 在异步计算并发射结果。