线程同步的方式【转】

引自：建议收藏！Kotlin 线程同步的 N 种方法

线程同步的方式

问题： 现有 Task1、Task2 等多个并行任务，如何等待全部任务执行完成后，开始执行 Task3 ?

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方法：让当前所在线程挂起，让指定线程先执行

@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：锁住一个对象，使其他线程无法访问该对象的同步代码块，但可以访问该对象的非同步代码块。

1. 修饰代码块：synchronized(this){}，锁住当前对象，作用范围是{}代码块
2. 修饰方法：锁住当前对象，作用范围是整个方法
3. 修饰静态方法：锁住当前类的所有对象，作用范围是整个静态方法
4. 修饰类：synchronized(XXX.class){}，锁住类的所有对象，作用范围是{}代码块

@Test
fun test_synchrnoized() {
	lateinit var s1: String
	lateinit var s2: String
	Thread {
		synchronized(Unit) {
			s1 =l task1()
		}
	}.start()
	s2 = task2()
  //task3在task2之后执行，task1执行完之后释放锁，才能执行task3
	synchronized(Unit) {
		task3(s1, s2)
	}
}

多个并行任务写法比较复杂，需要嵌套synchronized

ReentrantLock

ReentrantLock是JUC（java.util .concurrent，并发工具包）提供的线程锁，可以替换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()
}

多个并行任务需要创建多个Lock管理不同的任务

BlockingQueue

阻塞队列：阻塞队列内部也是通过 ReentrantLock 实现的。

阻塞队列更多是使用在生产/消费场景中的同步。

@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，它可以让一个线程一直处于阻塞状态，直到其他线程的执行全部完成：

共享锁的好处是不必为了每个任务都创建单独的锁，即使再多并行任务写起来也很轻松

@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 的命名由来，可以循环使用

@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 的原子类计数，达到实现无锁操作的目的。

@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

@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-- 并非线程安全，需要加锁操作

1. 获取cnt变量为2
2. 操作数-1
3. 写回内存

volatile只能保证读取的时候是最新的，当两个线程都读完之后，分别执行了2步骤，再写回内存，结果被覆盖

Future

上面无论有锁操作还是无锁操作，都需要定义两个变量s1、s2记录结果非常不方便。Java 1.5 开始，提供了 Callable 和 Future ，可以在任务执行结束时返回结果。

通过 future.get()，可以同步等待结果返回，写起来非常方便

@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 进行逻辑组合、实现复杂的异步编程。这些逻辑组合的方法以回调的形式避免了线程阻塞：

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

RxJava

RxJava 提供的各种操作符以及线程切换能力同样可以帮助我们实现需求：zip 操作符可以组合两个 Observable 的结果；subscribeOn 用来启动异步任务

@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特有

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

Flow

@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 在异步计算并发射结果。