RU
EN
ES
ZH
FR
DE
PT
AR
HI
Конкурентность и многопоточность в Kotlin: продвинутый уровень
В современной разработке эффективное управление конкурентностью — ключ к созданию быстрых и отзывчивых приложений. Kotlin, благодаря встроенной поддержке корутин (coroutines), предлагает мощный и гибкий инструментарий, выходящий далеко за рамки стандартных потоков Java. В этой статье мы рассмотрим продвинутые концепции конкурентности и многопоточности в Kotlin, включая работу с Channel, Flow, Mutex и тонкости планировщиков.
1. Корутины: глубже, чем основы
Корутины — это легковесные потоки, которые позволяют писать асинхронный код в императивном стиле. Однако на продвинутом уровне важно понимать, как работают диспетчеры (Dispatchers) и контексты корутин.
Планировщики (Dispatchers)
Выбор правильного диспетчера критичен для производительности:
Dispatchers.Default— для CPU-интенсивных задач (по умолчанию использует пул потоков, равный количеству ядер процессора).Dispatchers.IO— для операций ввода-вывода (сеть, файлы). Пул потоков может расширяться до 64 потоков (или больше, в зависимости от конфигурации).Dispatchers.Main— для UI-потока (Android, JavaFX).newSingleThreadContextиnewFixedThreadPoolContext— для создания собственных пулов.
import kotlinx.coroutines.*
fun main() = runBlocking {
launch(Dispatchers.Default) {
println("CPU-bound task on ${Thread.currentThread().name}")
}
launch(Dispatchers.IO) {
println("IO-bound task on ${Thread.currentThread().name}")
}
}Структурированная конкурентность
Одна из главных фич Kotlin — структурированная конкурентность. Она гарантирует, что дочерние корутины завершатся до завершения родительской. Это предотвращает утечки памяти и упрощает управление жизненным циклом.
fun main() = runBlocking {
coroutineScope {
launch {
delay(1000)
println("Task 1 completed")
}
launch {
delay(500)
println("Task 2 completed")
}
}
println("All tasks completed")
}2. Каналы (Channel) для передачи данных между корутинами
Channel — это аналог блокирующей очереди (BlockingQueue), но для корутин. Он позволяет безопасно передавать данные между корутинами, приостанавливая отправителя, если канал полон, или получателя, если канал пуст.
Типы каналов
- RendezvousChannel (по умолчанию) — не имеет буфера. Отправка блокируется до получения.
- BufferedChannel — имеет буфер заданного размера.
- ConflatedChannel — хранит только последнее отправленное значение, перезаписывая предыдущие.
- UnlimitedChannel — неограниченный буфер (осторожно: может вызвать OOM).
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
fun main() = runBlocking {
val channel = Channel<Int>(capacity = 5) // Buffered channel
launch {
for (i in 1..10) {
channel.send(i)
println("Sent: $i")
}
channel.close()
}
launch {
for (value in channel) {
println("Received: $value")
delay(100)
}
}
}Fan-out и Fan-in
Каналы отлично подходят для паттернов "веерный выход" (один продюсер, несколько консьюмеров) и "веерный вход" (несколько продюсеров, один консьюмер).
3. Flow: реактивные потоки данных
В отличие от Channel, который является "горячим" (данные отправляются независимо от наличия подписчика), Flow — это "холодный" поток, который начинает эмитировать данные только при наличии подписки. Flow идеален для обработки последовательностей асинхронных данных.
Создание Flow
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun simpleFlow(): Flow<Int> = flow {
for (i in 1..5) {
delay(100)
emit(i)
}
}
fun main() = runBlocking {
simpleFlow().collect { value ->
println("Received: $value")
}
}Продвинутые операторы
map,filter— стандартные операторы трансформации.flatMapConcat,flatMapMerge,flatMapLatest— для работы с вложенными потоками.buffer— для буферизации эмиссии.conflate— пропускать старые значения, если обработчик не успевает.catch— для обработки исключений внутри потока.
fun main() = runBlocking {
(1..10).asFlow()
.filter { it % 2 == 0 }
.map { it * it }
.buffer(5)
.collect { println(it) }
}4. Синхронизация: Mutex, Semaphore и атомарные операции
Несмотря на то, что корутины выполняются в одном потоке, при использовании нескольких потоков (например, через Dispatchers.Default) возникает необходимость в синхронизации доступа к общим ресурсам.
Mutex (взаимное исключение)
Mutex — это неблокирующий мьютекс для корутин. Он приостанавливает корутину, а не поток, что делает его очень эффективным.
import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
suspend fun main() = coroutineScope {
val mutex = Mutex()
var counter = 0
massiveRun {
mutex.withLock {
counter++
}
}
println("Counter = $counter")
}
suspend fun massiveRun(action: suspend () -> Unit) {
val jobs = List(1000) {
launch(Dispatchers.Default) {
repeat(1000) { action() }
}
}
jobs.forEach { it.join() }
}
Блоги
Рекомендаций по книгам
Читайте также
