Task Queues for Kotlin: Coroutines and KJobQueue

In modern Kotlin development — whether it's an Android app, a Ktor backend, or a microservice — there is often a need to execute background tasks asynchronously and in the correct order. A job queue helps manage such operations: file uploads, data synchronization, analytics sending, image processing. In this article, we will look at how to build a task queue using Kotlin coroutines, and also get acquainted with the KJobQueue library — a lightweight solution for advanced task management.

Why do you need a task queue?

The main problems that a task queue solves:

Execution order — tasks should not conflict over resources.
Concurrency limitation — for example, no more than 3 simultaneous downloads.
Error handling and retries — a task can be restarted upon failure.
Priorities — urgent tasks are executed first.
Restart resilience — the queue can persist its state.

Simple queue with coroutines (Channel)

The simplest way to implement a queue is to use a Channel and a handler coroutine. Example:

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

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

suspend fun processTask(task: Task) {    println("Processing task ${task.id}: ${task.payload}")    delay(500) // simulating work}

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

    // Launch a worker    val worker = launch {        for (task in taskChannel) {            processTask(task)        }    }

    // Add tasks    launch {        repeat(5) { i ->            taskChannel.send(Task(i, "Data $i"))        }        taskChannel.close()    }

    worker.join()}

This approach is suitable for simple scenarios but does not support priorities, retries, or state persistence.

KJobQueue — a library for an advanced queue

KJobQueue (available on GitHub) is a lightweight library for Kotlin built on top of coroutines. It provides:

Priority queue (based on Comparable).
Concurrency limitation (number of simultaneously executing tasks).
Automatic retries with delay.
Support for cancellation and shutdown.
Configurable dispatcher (thread pool).

Installation

Add to build.gradle.kts:

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

Example of using 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("Downloading $url (priority $priority)")        delay(1000)        if (Random.nextBoolean()) throw RuntimeException("Network error")    }}

fun main() = runBlocking {    val queue = JobQueue(        maxConcurrent = 2,          // maximum 2 tasks simultaneously        retryDelayMs = 500L,        // delay before retry        maxRetries = 3              // maximum 3 retries    )

    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) // give time for execution    queue.shutdown()}