Конкурентность и многопоточность в Kotlin: продвинутый уровень

Конкурентность и многопоточность в Kotlin: продвинутый уровень

В современной разработке на Kotlin, особенно для серверных и Android-приложений, грамотное управление конкурентностью — ключ к производительности и отзывчивости. На продвинутом уровне недостаточно просто запустить корутину. Нужно понимать, как эффективно синхронизировать доступ к данным, управлять потоками и избегать “адских” блокировок. В этой статье мы разберем продвинутые техники: от тонкостей kotlinx.coroutines до thread-safe коллекций и каналов.

1. Глубокое погружение в корутины: Structured Concurrency и Scopes

Основой продвинутой работы является Structured Concurrency. Это принцип, при котором жизненный цикл корутины привязан к области видимости (scope). Если родительская корутина отменяется, все её дочерние задачи автоматически завершаются. Это предотвращает утечки памяти и неконтролируемые процессы.

На продвинутом уровне вы должны создавать собственные скоупы, а не полагаться на глобальный GlobalScope.

import kotlinx.coroutines.*
import kotlin.coroutines.CoroutineContext

class DataRepository {
    // Собственный scope с диспетчером и обработчиком исключений
    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob() + CoroutineExceptionHandler { _, throwable ->
        println("Необработанная ошибка: ${throwable.message}")
    })

    fun loadData() {
        scope.launch {
            // дочерние корутины
            val data = async { fetchFromNetwork() }
            val cache = async { loadFromCache() }
            process(data.await(), cache.await())
        }
    }

    fun cleanup() {
        scope.cancel() // отменяем все корутины в скоупе
    }
}

Используйте SupervisorJob, когда сбой одной дочерней корутины не должен отменять остальные. Это критично для UI-слоев или параллельных независимых запросов.

2. Thread-Safe коллекции и атомарные операции

Даже с корутинами, если вы работаете с общими изменяемыми данными, нужна синхронизация. Kotlin предлагает несколько подходов:

• Атомарные типы (AtomicInteger, AtomicReference) — для простых счетчиков и флагов.
• Thread-Safe коллекции (ConcurrentHashMap, CopyOnWriteArrayList) — для сложных структур.
• Mutex из kotlinx.coroutines — для критических секций внутри корутин.

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger

class SafeCounter {
    private val mutex = Mutex()
    private var counter = 0

    suspend fun increment() {
        mutex.withLock {
            counter++ // критическая секция
        }
    }

    fun getCounter(): Int = counter
}

class CacheManager {
    private val cache = ConcurrentHashMap<String, String>()
    private val hitCount = AtomicInteger(0)

    fun put(key: String, value: String) {
        cache[key] = value
    }

    fun get(key: String): String? {
        val result = cache[key]
        if (result != null) hitCount.incrementAndGet()
        return result
    }
}

Важно: Mutex — suspend-функция, поэтому он не блокирует поток, а приостанавливает корутину. Это эффективнее, чем synchronized.

3. Каналы (Channels) и Flow: реактивные потоки данных

Для передачи данных между корутинами используйте Channels (аналог BlockingQueue для корутин) и Flow (холодный асинхронный поток).

Каналы: Producer-Consumer

import kotlinx.coroutines.channels.*
import kotlinx.coroutines.*

fun CoroutineScope.produceNumbers(): ReceiveChannel<Int> = produce {
    var x = 1
    while (x <= 10) {
        send(x++) // отправляем данные
        delay(100)
    }
}

fun CoroutineScope.processNumbers(channel: ReceiveChannel<Int>) = launch {
    for (number in channel) {
        println("Получено: $number")
    }
}

fun main() = runBlocking {
    val channel = produceNumbers()
    processNumbers(channel)
    delay(2000)
    channel.cancel() // закрываем канал
}

Flow: холодный асинхронный стрим

Flow идеален для обработки последовательностей данных (например, поток обновлений базы данных или результаты поиска). Он ленивый и поддерживает операторы как в коллекциях.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun searchFlow(query: String): Flow<String> = flow {
    emit("Начинаем поиск: $query")
    delay(500) // имитация сетевого запроса
    emit("Результат 1...")
    delay(300)
    emit("Результат 2...")
}.flowOn(Dispatchers.Default) // меняем контекст выполнения

fun main() = runBlocking {
    searchFlow("Kotlin")
        .catch { e -> emit("Ошибка: ${e.message}") }
        .collect { println(it) }
}

Продвинутый трюк: используйте flatMapMerge для параллельной обработки потоков, но контролируйте количество одновременных корутин через параметр concurrency.

4. Оптимизация диспетчеров и пулов потоков

Выбор правильного CoroutineDispatcher критичен для производительности:

• Dispatchers.Main — для UI (Android/Swing).
• Dispatchers.IO — для блокирующих операций (чтение файлов, сеть).
• Dispatchers.Default — для CPU-интенсивных задач.
• newSingleThreadContext() — для последовательного доступа к разделяемым ресурсам.

На продвинутом уровне вы можете создать свой пул потоков с ограничениями:

import kotlinx.coroutines.*
import java.util.concurrent.Executors

val customDispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()

fun main() = runBlocking {
    val jobs = List(100) {
        launch(customDispatcher) {
            // CPU-интенсивная работа
            var sum = 0L
            for (i in 1..1_