Работа с очередями задач в Kotlin: на практике

Введение: зачем нужны очереди задач в Kotlin?

Современные приложения всё чаще требуют параллельной обработки данных, фоновых операций и асинхронных вызовов. Без грамотного управления задачами ваш код может столкнуться с проблемами производительности, утечками памяти или «зависаниями» интерфейса. Очереди задач в Kotlin — это не просто абстракция, а мощный инструмент, который помогает организовать выполнение работы в нужном порядке, с контролем количества потоков и приоритетов.

В этой статье мы разберём практические подходы к реализации очередей задач на Kotlin: от классического ThreadPoolExecutor до корутин с их гибкими диспетчерами. Вы увидите, как применить эти техники в реальных проектах, и получите готовые примеры кода.

1. Классический подход: ThreadPoolExecutor и BlockingQueue

Стандартная библиотека Java (и Kotlin, как JVM-язык) предоставляет мощный класс ThreadPoolExecutor. Он управляет пулом потоков и использует внутреннюю блокирующую очередь для хранения ожидающих задач.

Создание очереди с фиксированным числом потоков

Наиболее простой случай — пул с фиксированным количеством потоков. Все задачи, которые не могут быть выполнены немедленно, помещаются в очередь LinkedBlockingQueue.

import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit

fun main() {
    val executor = Executors.newFixedThreadPool(3)

    for (i in 1..10) {
        executor.submit {
            println("Задача $i выполняется в потоке ${Thread.currentThread().name}")
            Thread.sleep(1000)
        }
    }

    executor.shutdown()
    executor.awaitTermination(1, TimeUnit.MINUTES)
    println("Все задачи завершены")
}

Этот код демонстрирует базовую очередь: задачи отправляются в пул, и если все потоки заняты, они ждут в очереди.

Приоритетная очередь: PriorityBlockingQueue

Иногда требуется выполнять срочные задачи раньше остальных. Для этого используем PriorityBlockingQueue с кастомным компаратором.

import java.util.concurrent.PriorityBlockingQueue
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit

data class PriorityTask(val priority: Int, val name: String) : Comparable<PriorityTask> {
    override fun compareTo(other: PriorityTask): Int = other.priority - this.priority
}

fun main() {
    val queue = PriorityBlockingQueue<Runnable>(10) { r1, r2 ->
        val t1 = (r1 as PriorityTaskWrapper).task
        val t2 = (r2 as PriorityTaskWrapper).task
        t1.compareTo(t2)
    }

    val executor = ThreadPoolExecutor(2, 4, 1, TimeUnit.SECONDS, queue)

    class PriorityTaskWrapper(val task: PriorityTask, val action: () -> Unit) : Runnable {
        override fun run() = action()
    }

    for (i in 1..5) {
        val priority = if (i % 2 == 0) 10 else 1
        executor.submit(PriorityTaskWrapper(PriorityTask(priority, "Задача $i")) {
            println("Выполняется: ${PriorityTask(priority, "Задача $i")}")
            Thread.sleep(500)
        })
    }

    executor.shutdown()
    executor.awaitTermination(1, TimeUnit.MINUTES)
}

Здесь задачи с priority = 10 будут выполняться раньше, чем с priority = 1, даже если они добавлены позже.

2. Очереди задач с помощью корутин (kotlinx.coroutines)

Современный Kotlin предлагает корутины — лёгковесные потоки, которые позволяют строить асинхронные очереди без прямого управления потоками. Основной инструмент — Channel и Dispatchers.

Простая очередь через Channel

Channel — это аналог блокирующей очереди, но для корутин. Он позволяет безопасно передавать задачи между корутинами.

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

suspend fun processTask(task: String) {
    delay(1000)
    println("Обработана задача: $task в ${Thread.currentThread().name}")
}

fun main() = runBlocking {
    val channel = Channel<String>(capacity = 10)

    // Продюсер — отправляет задачи
    val producer = launch {
        for (i in 1..10) {
            channel.send("Task $i")
            println("Отправлена задача $i")
        }
        channel.close() // сигнал, что задач больше не будет
    }

    // Консюмер — выполняет задачи последовательно
    val consumer = launch(Dispatchers.Default) {
        for (task in channel) {
            processTask(task)
        }
    }

    producer.join()
    consumer.join()
}

Этот паттерн гарантирует, что задачи будут обработаны в порядке поступления, и не требует явного управления потоками.

Параллельная обработка с Dispatchers

Если вам нужно обрабатывать несколько задач одновременно, используйте Dispatchers.Default или кастомный пул через newFixedThreadPoolContext.

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

suspend fun heavyTask(id: Int) {
    delay(2000)
    println("Тяжёлая задача $id завершена")
}

fun main() = runBlocking {
    val channel = Channel<Int>(capacity = 5)
    val dispatcher = newFixedThreadPoolContext(3, "WorkerPool")

    // Продюсер
    launch {
        for (i in 1..8) {
            channel.send(i)
        }
        channel.close()
    }

    // Несколько консюмеров
    repeat(3) { consumerId ->
        launch(dispatcher) {
            for (taskId in channel) {
                println("Consumer $consumerId взял задачу $taskId")
                heavyTask(taskId)
            }
        }
    }

    delay(10000) // ждём завершения всех задач
}

Здесь три корутины в пуле параллельно забирают задачи из общего канала, что даёт эффект очереди с фиксированным числом работников.

3. Продвинутый сценарий: очередь с приоритетом на корутинах

К сожалению, стандартный Channel не поддерживает приоритеты. Однако мы можем реализовать свою очередь на основе PriorityQueue и мьютекса.

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import java.util.PriorityQueue

class PriorityTaskQueue(private val capacity: Int = Int.MAX_VALUE) {
    private val queue = PriorityQueue<Pair<Int, () -> Unit>(compareBy { it.first })
    private val mutex = Mutex()
    private var isClosed = false

    suspend fun add(priority: Int, task: () -> Unit