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 funs 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.