Concurrent Data Fetcher

Simulate fetching data from multiple sources concurrently using coroutines and async/await.

Coroutinesasync/awaitDispatchers

Overview

Fetching data from three independent sources one after another wastes time for no reason — if each simulated network call takes a second, doing them sequentially costs three seconds, but nothing about source B depends on source A finishing first. Kotlin coroutines exist to make that concurrency easy to express without touching a raw `Thread`: a `suspend fun` marks a function as one that can pause without blocking the thread it runs on, and `async { }` launches work that runs concurrently with everything else inside the same `coroutineScope`, later joined back together with `.await()`.

By the end of this tutorial you will have a console application with `suspend fun` functions that simulate network calls using `delay()` (a suspending, non-blocking sleep), a sequential fetch for comparison, and a concurrent fetch built on `async`/`await` inside `coroutineScope { }` — plus a brief, concrete look at `Dispatchers.IO` versus `Dispatchers.Default` and why the choice actually matters once these simulated calls are replaced with real ones.

What You'll Build

A `data class FetchResult(source: String, data: String, durationMs: Long)` capturing one source's outcome.
A `suspend fun fetchFromSource(name: String)` that simulates a network call with `delay()`.
A `fetchSequentially()` function that awaits each source one after another, for timing comparison.
A `fetchConcurrently()` function using `async { }` to start every source at once, then `awaitAll()`.
A short explanation of `Dispatchers.IO` vs `Dispatchers.Default` and when each belongs.
A `runBlocking { }` entry point that times and prints both approaches side by side.

Prerequisites

Data classes and functions, as used throughout the earlier Kotlin projects.
Coroutine basics — what a `suspend fun` is and why calling one requires being inside a coroutine.
The `async`/`await` pattern for launching concurrent work and collecting its result.
The idea of a coroutine scope (`coroutineScope { }`, `runBlocking { }`) as a boundary that waits for its children to finish.
Basic familiarity with `Dispatchers` as "which pool of threads a coroutine's code actually runs on."

Project Structure

The whole program lives in a single file, `ConcurrentDataFetcher.kt`, containing `FetchResult`, a `suspend fun fetchFromSource()` that simulates one network call, `fetchSequentially()` and `fetchConcurrently()` as two contrasting strategies over the same set of sources, and `fun main()` wrapped in `runBlocking { }`. This project requires the `kotlinx-coroutines-core` library (Kotlin's coroutines are a library feature, not a language keyword built into the compiler, aside from the `suspend` modifier itself), so building it needs that dependency on the classpath — via Gradle's `implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0")` or an equivalent for your build tool.

Step 1: Define the Result Data Class

`FetchResult` records which source produced a result, what the (simulated) data was, and how long that individual fetch took — the `durationMs` field is what makes the sequential-versus-concurrent timing comparison in Step 6 concrete instead of just a claim.

// Captures the outcome of one simulated fetch: which source, what data came
// back, and how long that individual call took, so the later comparison
// between sequential and concurrent fetching has real numbers to show.
data class FetchResult(
    val source: String,     // Which source this result came from
    val data: String,       // The (simulated) payload returned
    val durationMs: Long    // How long this individual fetch took, for the timing comparison in Step 6
)

Step 2: Write a Suspend Function That Simulates a Fetch

The `suspend` modifier on `fetchFromSource()` is what makes it callable from inside a coroutine and, crucially, from inside `async { }` in Step 4 — a regular (non-`suspend`) function cannot call `delay()`, because `delay()` is itself a `suspend fun`. `delay()` is a *suspending* sleep: it pauses this coroutine without blocking the underlying thread, so other coroutines sharing that thread can keep running during the pause, which is exactly what makes concurrent fetching in Step 4 actually overlap in time rather than merely appearing to.

import kotlinx.coroutines.delay

// "suspend" means this function can pause (via delay()) without blocking the
// thread it runs on, and that it may only be called from a coroutine or
// another suspend function — never from ordinary synchronous code.
suspend fun fetchFromSource(name: String, simulatedLatencyMs: Long): FetchResult {
    val start = System.currentTimeMillis()
    delay(simulatedLatencyMs)                          // Non-blocking "sleep": frees the thread for other coroutines
    val elapsed = System.currentTimeMillis() - start
    return FetchResult(name, "data-from-$name", elapsed)
}

Step 3: Fetch One Source Sequentially

`fetchSequentially()` calls `fetchFromSource()` with plain, ordinary function-call syntax inside a `for` loop — no `async`, no `await`. Because `fetchFromSource` is a `suspend fun`, each call still suspends rather than blocks, but the loop does not move on to the next source until the current `delay()` has fully finished, so the sources are fetched one at a time, and the total time is the *sum* of every individual latency.

suspend fun fetchSequentially(sources: List<Pair<String, Long>>): List<FetchResult> {
    val results = mutableListOf<FetchResult>()
    for ((name, latency) in sources) {                  // Destructures each Pair<String, Long> into name and latency
        results.add(fetchFromSource(name, latency))      // Waits for this fetch to fully finish before starting the next
    }
    return results
}

Step 4: Fetch All Sources Concurrently With async/await

`fetchConcurrently()` is the whole point of the project: `coroutineScope { }` opens a scope that will not return until every coroutine launched inside it has completed, and inside that scope, `sources.map { async { fetchFromSource(...) } }` starts *every* fetch essentially at once — each `async { }` call returns immediately with a `Deferred<FetchResult>` (a handle to a result that is not ready yet) rather than waiting for the fetch to finish. `awaitAll()` then suspends until every one of those deferred results is ready, and the total time ends up close to the *slowest single fetch*, not the sum of all of them.

import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope

suspend fun fetchConcurrently(sources: List<Pair<String, Long>>): List<FetchResult> {
    return coroutineScope {                                    // Waits for every child coroutine below before returning
        val deferredResults = sources.map { (name, latency) ->  // map{} over sources, launching one async job per source
            async {                                               // Starts immediately, running concurrently with the others
                fetchFromSource(name, latency)
            }
        }
        deferredResults.awaitAll()                               // Suspends until every Deferred<FetchResult> has completed
    }
}

Example Usage

Click Run to see what this code prints.

Step 5: Choosing a Dispatcher

A `Dispatcher` decides which thread pool a coroutine's code actually runs on. `Dispatchers.IO` is backed by a large, elastic pool of threads specifically tuned for work that spends most of its time *waiting* — real network calls, file reads, database queries — so many `IO`-dispatched coroutines can be blocked-on-I/O at once without starving the app of threads. `Dispatchers.Default` is backed by a pool sized to the number of CPU cores, tuned instead for work that keeps the CPU genuinely busy — sorting a huge list, parsing, image processing. This project's `delay()` calls are not real I/O, so they do not block a thread at all regardless of dispatcher, but the moment `fetchFromSource()` is rewritten to make an actual HTTP call, wrapping that call in `withContext(Dispatchers.IO) { }` becomes the right choice, precisely because a real network call blocks the calling thread while it waits.

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext

// Illustrates where a real network call would use Dispatchers.IO — this
// project's delay() is not real I/O, but a genuine HTTP client call inside
// fetchFromSource() should be wrapped exactly like this in a real app.
suspend fun fetchFromSourceRealistic(name: String, simulatedLatencyMs: Long): FetchResult {
    return withContext(Dispatchers.IO) {   // Runs the block on IO's thread pool, tuned for blocking/waiting work
        val start = System.currentTimeMillis()
        delay(simulatedLatencyMs)            // Stand-in for a real network call that would block this thread
        val elapsed = System.currentTimeMillis() - start
        FetchResult(name, "data-from-$name", elapsed)
    }
}

Step 6: Build the Program Entry Point

`main()` itself cannot be `suspend`, so `runBlocking { }` is what bridges ordinary synchronous code into the coroutine world — it starts a coroutine and blocks the actual calling thread until everything inside it completes, which is appropriate exactly at a program's entry point (or in a test), and nowhere else in real application code. Timing both `fetchSequentially()` and `fetchConcurrently()` over the same three sources back to back makes the concurrency payoff visible directly in the sample run below.

import kotlinx.coroutines.runBlocking

fun main() = runBlocking {   // Bridges synchronous main() into the coroutine world; blocks this thread until done
    val sources = listOf("API-A" to 500L, "API-B" to 800L, "API-C" to 300L)

    println("===== SEQUENTIAL FETCH =====")
    val seqStart = System.currentTimeMillis()
    val sequentialResults = fetchSequentially(sources)
    val seqTotal = System.currentTimeMillis() - seqStart
    for (result in sequentialResults) {
        println("${result.source}: ${result.data} (${result.durationMs}ms)")
    }
    println("Total sequential time: ${seqTotal}ms")

    println("\n===== CONCURRENT FETCH =====")
    val conStart = System.currentTimeMillis()
    val concurrentResults = fetchConcurrently(sources)
    val conTotal = System.currentTimeMillis() - conStart
    for (result in concurrentResults) {
        println("${result.source}: ${result.data} (${result.durationMs}ms)")
    }
    println("Total concurrent time: ${conTotal}ms")
}

Complete Code

Here is the full program with every declaration assembled in the correct order, ready to save as `ConcurrentDataFetcher.kt`. It depends on `kotlinx-coroutines-core`, so it needs that library on the classpath — with Gradle, add `implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0")` and run with `./gradlew run`, or compile directly with `kotlinc` pointing `-cp` at the coroutines jar.

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext

data class FetchResult(
    val source: String,
    val data: String,
    val durationMs: Long
)

suspend fun fetchFromSource(name: String, simulatedLatencyMs: Long): FetchResult {
    val start = System.currentTimeMillis()
    delay(simulatedLatencyMs)
    val elapsed = System.currentTimeMillis() - start
    return FetchResult(name, "data-from-$name", elapsed)
}

suspend fun fetchSequentially(sources: List<Pair<String, Long>>): List<FetchResult> {
    val results = mutableListOf<FetchResult>()
    for ((name, latency) in sources) {
        results.add(fetchFromSource(name, latency))
    }
    return results
}

suspend fun fetchConcurrently(sources: List<Pair<String, Long>>): List<FetchResult> {
    return coroutineScope {
        val deferredResults = sources.map { (name, latency) ->
            async {
                fetchFromSource(name, latency)
            }
        }
        deferredResults.awaitAll()
    }
}

// Illustrates where a real network call would use Dispatchers.IO instead of the default dispatcher.
suspend fun fetchFromSourceRealistic(name: String, simulatedLatencyMs: Long): FetchResult {
    return withContext(Dispatchers.IO) {
        val start = System.currentTimeMillis()
        delay(simulatedLatencyMs)
        val elapsed = System.currentTimeMillis() - start
        FetchResult(name, "data-from-$name", elapsed)
    }
}

fun main() = runBlocking {
    val sources = listOf("API-A" to 500L, "API-B" to 800L, "API-C" to 300L)

    println("===== SEQUENTIAL FETCH =====")
    val seqStart = System.currentTimeMillis()
    val sequentialResults = fetchSequentially(sources)
    val seqTotal = System.currentTimeMillis() - seqStart
    for (result in sequentialResults) {
        println("${result.source}: ${result.data} (${result.durationMs}ms)")
    }
    println("Total sequential time: ${seqTotal}ms")

    println("\n===== CONCURRENT FETCH =====")
    val conStart = System.currentTimeMillis()
    val concurrentResults = fetchConcurrently(sources)
    val conTotal = System.currentTimeMillis() - conStart
    for (result in concurrentResults) {
        println("${result.source}: ${result.data} (${result.durationMs}ms)")
    }
    println("Total concurrent time: ${conTotal}ms")
}

Sample Run

Click Run to see what this code prints.

Extend This Project

Replace `fetchFromSource()`'s simulated `delay()` with a real HTTP call using Ktor's client, wrapped in `withContext(Dispatchers.IO)` as shown in Step 5.
Add a timeout with `withTimeoutOrNull(1000) { fetchFromSource(...) }` so one slow source cannot stall the whole batch indefinitely.
Wrap each `async { }` block in a `try`/`catch` (or use `supervisorScope`) so one failing source does not cancel the others.
Add a `retryWithBackoff()` suspend function that retries a failed fetch a fixed number of times with an increasing `delay()` between attempts.
Convert `fetchConcurrently()` to report progress as each source finishes, using a `Channel` or `Flow` instead of collecting everything with `awaitAll()` at once.

Summary

You built a concurrent data fetcher that shows, with real timing numbers, why `async`/`await` inside `coroutineScope { }` beats calling several `suspend fun`s one after another: the concurrent version finishes in roughly the time of its slowest source, not the sum of all of them. You also saw why `Dispatchers.IO` and `Dispatchers.Default` are not interchangeable — one pool is tuned for work that waits, the other for work that computes — a distinction that will matter the moment this project's simulated `delay()` calls are replaced with real network or disk I/O.