Mutex และ Semaphore สำหรับสถานะที่ใช้ร่วมกัน

ปัญหาสถานะที่ใช้ร่วมกัน

import kotlinx.coroutines.*
fun main() = runBlocking {
    var counter = 0
    val jobs = List(1000) {
        launch(Dispatchers.Default) { counter++ } // race condition!
    }
    jobs.forEach { it.join() }
    println("Expected 1000, got: $counter") // likely wrong
}

พื้นฐาน Mutex

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
fun main() = runBlocking {
    val mutex = Mutex()
    var counter = 0
    val jobs = List(1000) {
        launch(Dispatchers.Default) {
            mutex.withLock { counter++ }  // safe
        }
    }
    jobs.forEach { it.join() }
    println("Counter: $counter") // 1000
}

ส่วนขยาย withLock

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
fun main() = runBlocking {
    val mutex = Mutex()
    var list = mutableListOf<Int>()
    List(10) {
        launch {
            mutex.withLock {
                list.add(it)
            }
        }
    }.forEach { it.join() }
    println(list.sorted())
}

lock / unlock ด้วยตนเอง

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
fun main() = runBlocking {
    val mutex = Mutex()
    mutex.lock()
    try {
        println("Critical section")
    } finally {
        mutex.unlock()  // must always release
    }
}

Mutex ไม่เป็นแบบ Reentrant

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
fun main() = runBlocking {
    val mutex = Mutex()
    // This deadlocks:
    // mutex.withLock { mutex.withLock { println("never") } }
    mutex.withLock {
        println("Acquired once — safe")
    }
}

พื้นฐาน Semaphore

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
fun main() = runBlocking {
    val semaphore = Semaphore(3) // max 3 concurrent
    List(10) { i ->
        launch {
            semaphore.withPermit {
                println("Concurrent task $i"); delay(100)
            }
        }
    }.forEach { it.join() }
}

ส่วนขยาย withPermit

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
fun main() = runBlocking {
    val sem = Semaphore(2)
    val results = mutableListOf<String>()
    List(5) { i ->
        launch {
            sem.withPermit {
                delay(50)
                synchronized(results) { results.add("task-$i") }
            }
        }
    }.forEach { it.join() }
    println(results)
}

จำกัดอัตราด้วย Semaphore

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
val apiSemaphore = Semaphore(5) // max 5 concurrent requests
suspend fun callApi(id: Int): String {
    return apiSemaphore.withPermit {
        delay(100) // simulate network
        "Response-$id"
    }
}
fun main() = runBlocking {
    val results = (1..20).map { async { callApi(it) } }.awaitAll()
    println("Got ${results.size} responses")
}

Mutex กับ Semaphore

import kotlinx.coroutines.sync.*
// Mutex = Semaphore(1) for exclusive access
// Semaphore(n) = n-way concurrency limit
val exclusive = Mutex()          // one at a time
val limited   = Semaphore(3)     // three at a time
fun main() { println("Mutex for exclusion; Semaphore for rate limiting") }

ตัวนับแบบอะตอมิกด้วย Mutex

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
class AtomicCounter {
    private val mutex = Mutex()
    private var count = 0
    suspend fun increment() = mutex.withLock { count++ }
    suspend fun get() = mutex.withLock { count }
}
fun main() = runBlocking {
    val counter = AtomicCounter()
    List(100) { launch { counter.increment() } }.forEach { it.join() }
    println("Count: ${counter.get()}")
}

ทางเลือก: ตัวจัดส่งงานแบบเธรดเดียว

import kotlinx.coroutines.*
fun main() = runBlocking {
    val singleThread = newSingleThreadContext("CounterThread")
    var counter = 0
    List(1000) {
        launch(singleThread) { counter++ }
    }.forEach { it.join() }
    println("Counter: $counter") // 1000, no mutex needed
    singleThread.close()
}

ตรวจสอบความเข้าใจอย่างรวดเร็ว

สรุปทบทวน