Как можно реализовать синхронизацию в Kotlin с использованием корутин?

Синхронизация в корутинах Kotlin достигается с помощью специальных примитивов и структур данных из библиотеки kotlinx.coroutines.

Основные подходы:

1. Mutex

   • Предоставляет взаимное исключение для защиты критических секций кода.
   • Корутина, пытающаяся получить Mutex, блокируется до тех пор, пока он не освободится.
   • Имеет функции lock() и unlock(), а также более идиоматический withLock.

   import kotlinx.coroutines.sync.Mutex
   import kotlinx.coroutines.sync.withLock
   import kotlinx.coroutines.*
   
   val mutex = Mutex()
   var counter = 0
   
   suspend fun increment() {
       mutex.withLock {
           counter++
       }
   }
   
   suspend fun main() = coroutineScope {
       repeat(10_000) {
           launch {
               increment()
           }
       }
       println(counter) // Ожидаем 10000
   }
   

2. Semaphore

   • Ограничивает количество корутин, которые могут одновременно получить доступ к ресурсу или выполнить определенный блок кода.
   • Управляет пулом разрешений.
   • Имеет функции acquire() и release(), а также withPermit.

   import kotlinx.coroutines.sync.Semaphore
   import kotlinx.coroutines.sync.withPermit
   import kotlinx.coroutines.*
   
   val semaphore = Semaphore(2) // Одновременно могут работать 2 корутины
   
   suspend fun doLimitedWork(id: Int) {
       semaphore.withPermit {
           println("Coroutine $id acquired a permit. Working...")
           delay(100) // Имитация работы
           println("Coroutine $id released a permit.")
       }
   }
   
   suspend fun main() = coroutineScope {
       repeat(5) { i ->
           launch {
               doLimitedWork(i)
           }
       }
   }
   

3. Atomic operations (из kotlinx.coroutines.atomic)

   • Предоставляют потокобезопасные операции над примитивными типами и ссылками.
   • Используют низкоуровневые CPU-инструкции (CAS - Compare-and-Swap).
   • Подходят для простых операций без явных блокировок.

   import kotlinx.coroutines.atomic.AtomicInt
   import kotlinx.coroutines.*
   
   val atomicCounter = AtomicInt(0)
   
   suspend fun atomicIncrement() {
       atomicCounter.incrementAndGet()
   }
   
   suspend fun main() = coroutineScope {
       repeat(10_000) {
           launch {
               atomicIncrement()
           }
       }
       println(atomicCounter.value) // Ожидаем 10000
   }
   

4. Shared data structures on single-threaded dispatcher

   • Самый простой подход. Запуск корутин на однопоточном контексте (Dispatchers.Default.limitedParallelism(1) или newSingleThreadContext) гарантирует последовательное выполнение кода, исключая гонки данных. Не является явным примитивом синхронизации, но обеспечивает синхронизацию путем сериализации доступа.

   import kotlinx.coroutines.*
   
   val mySingleThreadContext = newSingleThreadContext("SingleThread")
   var sharedData = mutableListOf<Int>()
   
   suspend fun addToSharedData(value: Int) {
       sharedData.add(value) // Безопасно, так как выполняется на одном потоке
   }
   
   suspend fun main() = withContext(mySingleThreadContext) {
       repeat(10_000) {
           launch {
                addToSharedData(it)
           }
       }
       println(sharedData.size) // Ожидаем 10000
   }
   

5. Channels

   • Не являются примитивом синхронизации в чистом виде, но могут использоваться для безопасной передачи данных между корутинами, что косвенно решает проблемы синхронизации доступа к передаваемым данным.
   • Предоставляют способ передачи потока данных из одной корутины в другую.

Выбор подхода зависит от сценария использования. Mutex и Semaphore предоставляют классические механизмы блокировки, Atomic операции эффективны для простых атомарных обновлений, а однопоточный диспетчер удобен, когда доступ к общим данным должен быть строго последовательным. Channels используются для коммуникации и координации, а не прямой защиты общих ресурсов.