Очереди задач в Kotlin: создание, управление и оптимизация

Введение в очереди задач в Kotlin

В современной разработке программного обеспечения часто возникает необходимость обрабатывать задачи асинхронно, особенно при работе с сетевыми запросами, операциями ввода-вывода или фоновыми вычислениями. Очереди задач (task queues) — это один из ключевых паттернов, который позволяет организовать упорядоченное выполнение заданий, избегая блокировки основного потока. В языке Kotlin, благодаря его тесной интеграции с JVM и нативной поддержке корутин, реализация очередей становится элегантной и эффективной.

В этой статье мы рассмотрим, как создавать и управлять очередями задач в Kotlin, используя как классические подходы (на основе потоков), так и современные возможности корутин. Вы узнаете о различных типах очередей, их преимуществах и получите практические примеры кода, которые можно сразу применить в своих проектах.

Основные подходы к реализации очередей задач

В Kotlin существует несколько способов реализации очередей задач. Выбор подхода зависит от требований к производительности, сложности и используемой среды.

1. Очереди на основе потоков (BlockingQueue)

Классический способ, который используется в Java и полностью совместим с Kotlin. BlockingQueue из пакета java.util.concurrent предоставляет потокобезопасную очередь с блокирующими операциями. Этот подход хорошо подходит для многопоточных приложений, где несколько воркеров обрабатывают задачи.

import java.util.concurrent.BlockingQueue
import java.util.concurrent.LinkedBlockingQueue

class ThreadBasedTaskQueue {
    private val queue: BlockingQueue<Runnable> = LinkedBlockingQueue()
    private val workers: List<Thread>

    constructor(workerCount: Int) {
        workers = (1..workerCount).map { workerId ->
            Thread {
                while (true) {
                    val task = queue.take() // блокируется, пока нет задач
                    try {
                        task.run()
                    } catch (e: Exception) {
                        println("Worker $workerId error: ${e.message}")
                    }
                }
            }.apply { start() }
        }
    }

    fun submit(task: Runnable) {
        queue.put(task)
    }
}

// Использование
fun main() {
    val taskQueue = ThreadBasedTaskQueue(4)
    repeat(10) { index ->
        taskQueue.submit {
            println("Task $index executed by ${Thread.currentThread().name}")
        }
    }
    Thread.sleep(1000)
}

Преимущества: простота, предсказуемое поведение, хорошая изоляция задач. Недостатки: накладные расходы на создание потоков, ограниченное количество потоков в системе.

2. Очереди на основе корутин (Channel)

Kotlin Coroutines предлагает более легковесный механизм — Channel. Каналы позволяют передавать данные между корутинами без блокировки потоков. Это идеальное решение для асинхронных приложений.

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

class CoroutineTaskQueue(private val workerCount: Int = 4) {
    private val channel = Channel<suspend () -> Unit>(Channel.UNLIMITED)

    fun start(scope: CoroutineScope) {
        repeat(workerCount) { workerId ->
            scope.launch {
                for (task in channel) {
                    try {
                        task()
                    } catch (e: Exception) {
                        println("Coroutine worker $workerId error: ${e.message}")
                    }
                }
            }
        }
    }

    suspend fun submit(task: suspend () -> Unit) {
        channel.send(task)
    }
}

// Использование
fun main() = runBlocking {
    val taskQueue = CoroutineTaskQueue(4)
    taskQueue.start(this)

    repeat(10) { index ->
        taskQueue.submit {
            delay(100)
            println("Task $index completed on ${Thread.currentThread().name}")
        }
    }
    delay(2000) // ждем завершения
}

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

Продвинутые техники управления очередями

После освоения базовых подходов перейдем к более сложным сценариям: обработка ошибок, приоритезация задач и динамическое масштабирование.

3. Очередь с приоритетами и обработкой ошибок

В реальных проектах задачи могут иметь разный приоритет. Kotlin позволяет легко реализовать очередь с приоритетами, используя PriorityQueue или кастомные корутины. Также важно корректно обрабатывать ошибки, чтобы одна упавшая задача не блокировала всю очередь.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import java.util.PriorityQueue

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

class PriorityTaskQueue(private val workerCount: Int = 2) {
    private val channel = Channel<PriorityTask>(Channel.UNLIMITED)

    fun start(scope: CoroutineScope) {
        repeat(workerCount) {
            scope.launch {
                for (task in channel) {
                    try {
                        withTimeout(5000) { task.action() }
                    } catch (e: TimeoutCancellationException) {
                        println("Task timed out, resubmitting with lower priority")
                        channel.send(task.copy(priority = task.priority - 1))
                    } catch (e: Exception) {
                        println("Task failed: ${e.message}")
                    }
                }
            }
        }
    }

    suspend fun submit(priority: Int = 0, action: suspend () -> Unit) {
        channel.send(PriorityTask(priority, action))
    }
}

// Пример использования
fun main() = runBlocking {
    val queue = PriorityTaskQueue(2)
    queue.start(this)

    queue.submit(priority = 1) {
        delay(100)
        println("Low priority task")
    }
    queue.submit(priority = 10) {
        println("High priority task")
    }
    queue.submit(priority = 5) {
        delay(2000)
        println("Medium priority task with delay")
    }
    delay(3000)
}

В этом примере задачи с более высоким приоритетом выполняются раньше. Если задача не укладывается в таймаут, она автоматически переносится в конец очереди с пониженным приоритетом.

4. Динамическое масштабирование воркеров

Иногда количество задач резко возрастает, и статическое количество воркеров не справляется. Решением может быть динамическое добавление или удаление воркеров в зависимости от нагрузки.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.collect

class DynamicTaskQueue(private val minWorkers: Int = 2, private val maxWorkers: Int = 10) {
    private val channel = Channel<suspend ()