Kotlin 协程的并发问题

Kotlin 协程的并发问题,第1张

文章目录 Kotlin 协程并发问题协程并发问题使用synchronized解决协程并发问题使用单线程解决并发问题使用Mutex(协程同步锁)解决并发问题withLock 使用Actor并发同步模型避免共享变量

Kotlin 协程的并发问题 协程并发问题
fun main() = runBlocking {
    var i = 0
    val jobs = mutableListOf<Job>()

    repeat(10) {
        val job = launch(Dispatchers.Default) {
            repeat(1000) {
                i++
            }
        }
        jobs.add(job)
    }

    jobs.joinAll()

    println("i = $i")
}
/*
输出信息:
i = 9703
 */

说明:创建了10个协程,每个协程对i进行1000次自增 *** 作,正常情况下结果应该是10000,但是实际上大概率不会是10000。这是因为这10个协程分别运行在不同的线程中,出现了并发同步问题。

使用synchronized解决协程并发问题

Kotlin是基于JVM的,所以可以使用Java中的同步手段,如synchronized、Lock、Atomic等。

但是sychronized是线程模型下的,不支持协程中的挂起函数。

fun main() = runBlocking {
    val lock = Any()
    var i = 0
    val jobs = mutableListOf<Job>()

    repeat(10) {
        val job = launch(Dispatchers.Default) {
            repeat(1000) {
                synchronized(lock) {
                    i++
                }
            }
        }
        jobs.add(job)
    }

    jobs.joinAll()

    println("i = $i")
}
/*
输出信息:
i = 10000
 */
fun main() = runBlocking {
    var i = 0
    val jobs = mutableListOf<Job>()

    @Synchronized
    fun add() {
        i++
    }

    repeat(10) {
        val job = launch(Dispatchers.Default) {
            repeat(1000) {
                add()
            }
        }
        jobs.add(job)
    }

    jobs.joinAll()

    println("i = $i")
}

/*
输出信息:
i = 10000
 */
使用单线程解决并发问题
fun main() = runBlocking {
    val mySingleDispatcher = Executors.newSingleThreadExecutor {
        Thread(it, "我的线程").apply { isDaemon = true }
    }.asCoroutineDispatcher()

    var i = 0
    val jobs = mutableListOf<Job>()

    repeat(10) {
        val job = launch(mySingleDispatcher) {
            repeat(1000) {
                i++
            }
        }
        jobs.add(job)
    }

    jobs.joinAll()

    println("i = $i")
}

/*
输出信息:
i = 10000
 */

使用launch(mySingleDispatcher)后,所有的协程任务都运行在单线程mySingleDispatcher中,这样就没有并发问题了。

使用Mutex(协程同步锁)解决并发问题

在Java中,锁是阻塞式的,会大大影响协程的非阻塞式的特性,所以不推荐使用传统的同步锁。

在Kotlin协程中,可以使用Mutex非阻塞式锁。Mutex#lock()是一个挂起函数,这是实现非阻塞式同步锁的根本原因。

fun main() = runBlocking {
    val mutex = Mutex()
    var i = 0
    val jobs = mutableListOf<Job>()

    repeat(10) {
        val job = launch(mySingleDispatcher) {
            repeat(1000) {
                mutex.lock()
                try {
                    i++
                } finally {
                    mutex.unlock()
                }
            }
        }
        jobs.add(job)
    }

    jobs.joinAll()

    println("i = $i")
}

/*
输出信息:
i = 10000
 */
withLock

withLock扩展函数的本质,其实是在 finally{} 当中调用了 unlock()。

fun main() = runBlocking {
    val mutex = Mutex()
    var i = 0
    val jobs = mutableListOf<Job>()

    repeat(10) {
        val job = launch(mySingleDispatcher) {
            repeat(1000) {
                mutex.withLock {
                    i++
                }
            }
        }
        jobs.add(job)
    }

    jobs.joinAll()

    println("i = $i")
}

/*
输出信息:
i = 10000
 */
使用Actor并发同步模型

Actor是一个并发同步模型,本质是基于Channel管道消息实现的。

sealed class Msg {
    object AddMsg : Msg()
    class ResultMsg(val result: CompletableDeferred<Int>) : Msg()
}

fun main() = runBlocking {
    suspend fun addActor() = actor<Msg> {
        var i = 0
        for (msg in channel) {
            when (msg) {
                is Msg.AddMsg -> i++
                is Msg.ResultMsg -> msg.result.complete(i)
            }
        }
    }

    val actor = addActor()
    val jobs = mutableListOf<Job>()

    repeat(10) {
        val job = launch(mySingleDispatcher) {
            repeat(1000) {
                actor.send(Msg.AddMsg)
            }
        }
        jobs.add(job)
    }

    jobs.joinAll()

    val deferred = CompletableDeferred<Int>()
    actor.send(Msg.ResultMsg(deferred))
    val result = deferred.await()
    actor.close()
    
    println("i = $result")
}

/*
输出信息:
i = 10000
 */
避免共享变量
fun main() = runBlocking {
    val deferreds = mutableListOf<Deferred<Int>>()

    repeat(10) {
        val deferred = async(Dispatchers.Default) {
            var i = 0
            repeat(1000) {
                i++
            }
            return@async i
        }
        deferreds.add(deferred)
    }

    var result = 0
    deferreds.forEach {
        result += it.await()
    }
    
    println("i = $result")
}

/*
输出信息:
i = 10000
 */

欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/web/990105.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2022-05-21
下一篇 2022-05-21

发表评论

登录后才能评论

评论列表(0条)

保存