面试必备:Kotlin 线程同步的 N 种方法

简介: 面试的时候经常会被问及多线程同步的问题,例如,有 Task1、Task2 等多个并行任务,如何等待全部执行完成后执行 Task3?

面试的时候经常会被问及多线程同步的问题,例如:

“ 现有 Task1、Task2 等多个并行任务,如何等待全部执行完成后,执行 Task3。”

在 Kotlin 中我们有多种实现方式,本文将所有这些方式做了整理,建议收藏。

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

我们先定义三个Task,模拟上述场景, Task3 基于 Task1、Task2 返回的结果拼接字符串,每个 Task 通过 sleep 模拟耗时:
image.png

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

1. Thread.join()

Kotlin 兼容 Java,Java 的所有线程工具默认都可以使用。其中最简单的线程同步方式就是使用 Threadjoin()

@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

使用 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

3. ReentrantLock

ReentrantLock 是 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()

    }

ReentrantLock 的好处是,当有多个并行任务时是不会出现嵌套 synchrnoized 的问题,但仍然需要创建多个 lock 管理不同的任务,

4. 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)
    }

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

5. 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)
    }

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

6. 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)

    }

7. 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

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

8. Future

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

@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(),可以同步等待结果返回,写起来非常方便

9. 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()
}

<br/>

10. 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()

}

11. Coroutine

前面讲了那么多,其实都是 Java 的工具。 Coroutine 终于算得上是 Kotlin 特有的工具了:

@Test
fun test_coroutine() {

    runBlocking {
        val c1 = async(Dispatchers.IO) {
            task1()
        }

        val c2 = async(Dispatchers.IO) {
            task2()
        }

        task3(c1.await(), c2.await())
    }
}

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

12. Flow

Flow 就是 Coroutine 版的 RxJava,具备很多 RxJava 的操作符,例如 zip:


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

总结

上面这么多方式,就像茴香豆的“茴”字的四种写法,没必要都掌握。作为结论,在 Kotlin 上最好用的线程同步方案首推协程!

目录
相关文章
|
3月前
|
人工智能 前端开发 Java
Java 面试资料中相关代码使用方法与组件封装方法解析
这是一份详尽的Java面试资料代码指南,涵盖使用方法与组件封装技巧。内容包括环境准备(JDK 8+、Maven/Gradle)、核心类示例(问题管理、学习进度跟踪)、Web应用部署(Spring Boot、前端框架)、单元测试及API封装。通过问题库管理、数据访问组件、学习进度服务和REST接口等模块化设计,帮助开发者高效组织与复用功能,同时支持扩展如用户认证、AI推荐等功能。适用于Java核心技术学习与面试备考,提升编程与设计能力。资源链接:[点此下载](https://pan.quark.cn/s/14fcf913bae6)。
87 6
Java 面试资料中相关代码使用方法与组件封装方法解析
|
8月前
|
监控 Kubernetes Java
阿里面试:5000qps访问一个500ms的接口,如何设计线程池的核心线程数、最大线程数? 需要多少台机器?
本文由40岁老架构师尼恩撰写,针对一线互联网企业的高频面试题“如何确定系统的最佳线程数”进行系统化梳理。文章详细介绍了线程池设计的三个核心步骤:理论预估、压测验证和监控调整,并结合实际案例(5000qps、500ms响应时间、4核8G机器)给出具体参数设置建议。此外,还提供了《尼恩Java面试宝典PDF》等资源,帮助读者提升技术能力,顺利通过大厂面试。关注【技术自由圈】公众号,回复“领电子书”获取更多学习资料。
|
5月前
|
人工智能 算法 数据库
美团面试:LLM大模型存在哪些问题?RAG 优化有哪些方法?_
美团面试:LLM大模型存在哪些问题?RAG 优化有哪些方法?_
|
8月前
|
安全 Java 程序员
面试必看:如何设计一个可以优雅停止的线程?
嘿,大家好!我是小米。今天分享一篇关于“如何停止一个正在运行的线程”的面试干货。通过一次Java面试经历,我明白了停止线程不仅仅是技术问题,更是设计问题。Thread.stop()已被弃用,推荐使用Thread.interrupt()、标志位或ExecutorService来优雅地停止线程,避免资源泄漏和数据不一致。希望这篇文章能帮助你更好地理解Java多线程机制,面试顺利! 我是小米,喜欢分享技术的29岁程序员。欢迎关注我的微信公众号“软件求生”,获取更多技术干货!
202 53
|
7月前
|
数据采集 Java Linux
面试大神教你:如何巧妙回答线程优先级这个经典考题?
大家好,我是小米。本文通过故事讲解Java面试中常见的线程优先级问题。小明和小华的故事帮助理解线程优先级:高优先级线程更可能被调度执行,但并非越高越好。实际开发需权衡业务需求,合理设置优先级。掌握线程优先级不仅能写出高效代码,还能在面试中脱颖而出。最后,小张因深入分析成功拿下Offer。希望这篇文章能助你在面试中游刃有余!
112 4
面试大神教你:如何巧妙回答线程优先级这个经典考题?
|
7月前
|
Java 程序员 开发者
Java社招面试题:一个线程运行时发生异常会怎样?
大家好,我是小米。今天分享一个经典的 Java 面试题:线程运行时发生异常,程序会怎样处理?此问题考察 Java 线程和异常处理机制的理解。线程发生异常,默认会导致线程终止,但可以通过 try-catch 捕获并处理,避免影响其他线程。未捕获的异常可通过 Thread.UncaughtExceptionHandler 处理。线程池中的异常会被自动处理,不影响任务执行。希望这篇文章能帮助你深入理解 Java 线程异常处理机制,为面试做好准备。如果你觉得有帮助,欢迎收藏、转发!
406 14
|
7月前
|
安全 Java 程序员
Java 面试必问!线程构造方法和静态块的执行线程到底是谁?
大家好,我是小米。今天聊聊Java多线程面试题:线程类的构造方法和静态块是由哪个线程调用的?构造方法由创建线程实例的主线程调用,静态块在类加载时由主线程调用。理解这些细节有助于掌握Java多线程机制。下期再见! 简介: 本文通过一个常见的Java多线程面试题,详细讲解了线程类的构造方法和静态块是由哪个线程调用的。构造方法由创建线程实例的主线程调用,静态块在类加载时由主线程调用。理解这些细节对掌握Java多线程编程至关重要。
124 13
|
7月前
|
缓存 安全 Java
面试中的难题:线程异步执行后如何共享数据?
本文通过一个面试故事,详细讲解了Java中线程内部开启异步操作后如何安全地共享数据。介绍了异步操作的基本概念及常见实现方式(如CompletableFuture、ExecutorService),并重点探讨了volatile关键字、CountDownLatch和CompletableFuture等工具在线程间数据共享中的应用,帮助读者理解线程安全和内存可见性问题。通过这些方法,可以有效解决多线程环境下的数据共享挑战,提升编程效率和代码健壮性。
221 6
|
8月前
|
算法 安全 Java
Java线程调度揭秘:从算法到策略,让你面试稳赢!
在社招面试中,关于线程调度和同步的相关问题常常让人感到棘手。今天,我们将深入解析Java中的线程调度算法、调度策略,探讨线程调度器、时间分片的工作原理,并带你了解常见的线程同步方法。让我们一起破解这些面试难题,提升你的Java并发编程技能!
239 16
|
8月前
|
Java 程序员 调度
Java 高级面试技巧:yield() 与 sleep() 方法的使用场景和区别
本文详细解析了 Java 中 `Thread` 类的 `yield()` 和 `sleep()` 方法,解释了它们的作用、区别及为什么是静态方法。`yield()` 让当前线程释放 CPU 时间片,给其他同等优先级线程运行机会,但不保证暂停;`sleep()` 则让线程进入休眠状态,指定时间后继续执行。两者都是静态方法,因为它们影响线程调度机制而非单一线程行为。这些知识点在面试中常被提及,掌握它们有助于更好地应对多线程编程问题。
299 9