【面试必知】Kotlin并行任务/线程同步的几种实现方式

阳光穿透心脏的1/2处 2022-12-06 05:29 140阅读 0赞

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ZpdGF2aXZh_size_16_color_FFFFFF_t_70_pic_center]

多线程开发中，经常会遇到这样的需求：  
Task1、Task2 等多个任务并行执行，等待全部执行完成后，执行Task3。  
我们在Kotlin种可以有多种方法实现，在这里一并列举一下：

1.  Thread.join
2.  Synchronized
3.  ReentrantLock
4.  BlockingQueue
5.  CountDownLatch
6.  CyclicBarrier
7.  CAS
8.  Future
9.  CompletableFuture
10. Rxjava
11. Coroutine

# Tasks Mock #

--------------------

Task3基于 Task1、Task2返回的结果，拼接字符串，每个任务都通过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") }
    }

# Thread.join() #

--------------------

Kotlin兼容Java，所以Java的所有工具都可以使用。本case中最简单的方式就是使用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)
    
    }

# Synchronized #

--------------------

使用`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)
            }
    
        }

但是如果超过三个任务，使用synchrnoized这种写法就比较别扭了，为了同步多个并行任务的结果需要声明n个锁，并且嵌套n个synchronized。

# ReentrantLock #

--------------------

Lock是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()
    
        }

当有更多任务时，不会出现嵌套synchrnoized的问题，但仍然需要创建多个lock管理不同的任务，

# BlockingQueue #

--------------------

阻塞队列内部也是通过Lock实现的，所以同样可以达到同步锁的效果

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

阻塞队列更多的用在生产/消费的任务模型中，用在这种case中只是为了凑数而已

# CountDownLatch #

--------------------

JUC中的锁大都是基于`AQS`实现的，包括独享锁和共享锁，例如ReentrantLock就是是一种独享锁。  
相比之下，共享锁更适合本case，例如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内部的通过自旋锁实现了同步，自旋锁是通过CAS避免了线程阻塞的开销。  
因此，我们可以使用基于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`循环空转看起来很浪费，但是自旋锁的本质就是这样，所以仅仅是用于一些cpu密集型的短任务同步。

# volatile #

--------------------

看到CAS，另外可能很多人会想,使用volatile也可以抱枕线程安全实现无锁操作呢? 如下

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

# Future #

--------------------

上面无论有锁操作还是无锁操作，都需要定义两个变量`s1`、`s2`记录结果非常不方便。  
Java 1.5开始，提供了Callable和Future，通过它们可以在任务执行完毕之后得到任务执行结果。

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

通过`future.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 #

--------------------

前面讲了那么多，其实都是Java的工具。  
最后一个Coroutine终于对得起本文标题了。

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

写起来特别舒服，可以说是集前面各类工具的优点于一身。

# 总结 #

--------------------

本文是列举了茴香豆的回字的所有四种写法，但是并非鼓励大家滥用。  
作为结论：Kotlin上最优雅最好用的并行任务实现方式首推协程！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ZpdGF2aXZh_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/f1a84339ccd74502946d8498bc3f894f.png