Очереди задач для Kotlin: Coroutines и KJobQueue

Очереди задач для Kotlin: Coroutines и KJobQueue

В современной разработке на Kotlin — будь то Android-приложение, бэкенд на Ktor или микросервис — часто возникает необходимость выполнять фоновые задачи асинхронно и в правильном порядке. Очередь задач (job queue) помогает управлять такими операциями: загрузка файлов, синхронизация данных, отправка аналитики, обработка изображений. В этой статье мы разберём, как построить очередь задач с помощью корутин Kotlin, а также познакомимся с библиотекой KJobQueue — лёгким решением для продвинутого управления задачами.

Зачем нужна очередь задач?

Основные проблемы, которые решает очередь задач:

• Порядок выполнения — задачи не должны конфликтовать за ресурсы.
• Ограничение параллелизма — например, не более 3 одновременных загрузок.
• Обработка ошибок и повторные попытки — при сбое задача может быть перезапущена.
• Приоритеты — срочные задачи выполняются раньше.
• Устойчивость к перезапуску — очередь может сохранять состояние.

Простая очередь на корутинах (Channel)

Самый простой способ реализовать очередь — использовать Channel и корутину-обработчик. Пример:

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

data class Task(val id: Int, val payload: String)

suspend fun processTask(task: Task) {
    println("Обработка задачи ${task.id}: ${task.payload}")
    delay(500) // имитация работы
}

fun main() = runBlocking {
    val taskChannel = Channel<Task>(capacity = 10)

    // Запускаем воркер
    val worker = launch {
        for (task in taskChannel) {
            processTask(task)
        }
    }

    // Добавляем задачи
    launch {
        repeat(5) { i ->
            taskChannel.send(Task(i, "Данные $i"))
        }
        taskChannel.close()
    }

    worker.join()
}

Этот подход подходит для простых сценариев, но не поддерживает приоритеты, повторы и сохранение состояния.

KJobQueue — библиотека для продвинутой очереди

KJobQueue (доступна на GitHub) — это лёгкая библиотека для Kotlin, построенная поверх корутин. Она предоставляет:

• Очередь с приоритетами (на основе Comparable).
• Ограничение параллелизма (количество одновременно выполняемых задач).
• Автоматические повторные попытки с задержкой.
• Поддержку отмены и остановки.
• Настраиваемый диспетчер (пул потоков).

Установка

Добавьте в build.gradle.kts:

dependencies {
    implementation("io.github.kjobqueue:kjobqueue-core:1.0.0")
}

Пример использования KJobQueue

import io.github.kjobqueue.*
import kotlinx.coroutines.*
import kotlin.random.Random

class DownloadJob(val url: String, override val priority: Int = 0) : PriorityJob {
    override suspend fun execute() {
        println("Скачивание $url (приоритет $priority)")
        delay(1000)
        if (Random.nextBoolean()) throw RuntimeException("Ошибка сети")
    }
}

fun main() = runBlocking {
    val queue = JobQueue(
        maxConcurrent = 2,          // максимум 2 задачи одновременно
        retryDelayMs = 500L,        // задержка перед повтором
        maxRetries = 3              // максимум 3 повтора
    )

    queue.add(DownloadJob("https://example.com/file1", priority = 5))
    queue.add(DownloadJob("https://example.com/file2", priority = 1))
    queue.add(DownloadJob("https://example.com/file3", priority = 10))

    delay(5000) // даём время на выполнение
    queue.shutdown()
}

Обработка ошибок и повторные попытки

KJobQueue автоматически обрабатывает исключения. Если задача выбросила исключение, она будет помещена обратно в очередь (с учётом maxRetries). Вы также можете подписаться на события:

queue.onJobFailed { job, throwable ->
    println("Задача ${job} провалилась: ${throwable.message}")
}

queue.onJobCompleted { job ->
    println("Задача ${job} выполнена успешно")
}

Сохранение очереди между перезапусками

Для долгоживущих приложений (сервер, Android-сервис) можно сохранять состояние очереди в БД. KJobQueue поддерживает плагины для хранения — например, kjobqueue-storage-sqlite:

val queue = Job