Kotlin 任务队列：协程与 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)

    // 启动 worker    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()}