Kotlin Academy · Lesson

flowOn and buffer: Context and Backpressure

Change emission context with flowOn and buffer emissions for backpressure.

Lesson 4 of 413 steps

flowOn and buffer: Context and Backpressure is a free Kotlin Academy lesson on CoddyKit. This is lesson 4 of 4. Any 3 lessons of this learning path are free to read in full — after that, CoddyKit PRO unlocks every lesson, plus hands-on practice with a built-in code editor and a 24/7 AI tutor. It is part of the Kotlin Academy learning path, and your progress syncs across the web and the CoddyKit app. The Kotlin Academy course includes 4 lessons in total.

Flow Context

By default, a flow runs in the context of the coroutine that calls collect. The dispatcher of the producer and consumer are the same unless you change it.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit on: ${Thread.currentThread().name}")
        emit(1)
    }.collect {
        println("collect on: ${Thread.currentThread().name}")
    }
}

flowOn Changes Upstream Context

flowOn(dispatcher) runs the upstream flow (everything above it in the chain) on the specified dispatcher, while collection stays on the caller's dispatcher.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit: ${Thread.currentThread().name}")
        emit(1)
    }.map {
        println("map: ${Thread.currentThread().name}")
        it * 2
    }.flowOn(Dispatchers.Default)  // above runs on Default
    .collect {
        println("collect: ${Thread.currentThread().name}")
    }
}

Multiple flowOn in a Chain

You can use flowOn multiple times. Each affects the operators directly above it up to the previous flowOn.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow { emit(1) }
        .map { it + 1 }.flowOn(Dispatchers.IO)      // map runs on IO
        .map { it * 2 }.flowOn(Dispatchers.Default)  // this map runs on Default
        .collect { println("Result: $it") }           // collect on Main (runBlocking)
}

The Backpressure Problem

When the producer emits faster than the collector can process, values queue up. Without buffering this causes the producer to wait.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i ->
            delay(100)  // fast producer
            emit(i)
        }
    }.collect {
        delay(300)      // slow consumer
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

buffer() Operator

buffer() runs the producer and consumer concurrently in separate coroutines, buffering emitted values in a channel. The producer does not wait for the consumer.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i -> delay(100); emit(i) }
    }.buffer()  // producer and consumer run concurrently
    .collect {
        delay(300)
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

buffer Capacity

buffer(capacity) sets the channel buffer size. When the buffer is full, the producer suspends (backpressure). Default capacity is 64.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
    flow { repeat(5) { emit(it) } }
        .buffer(Channel.RENDEZVOUS)    // 0: producer waits
        // .buffer(Channel.BUFFERED)   // default: 64
        // .buffer(Channel.UNLIMITED)  // unbounded
        .collect { delay(50); println(it) }
}

conflate() for Latest-Only

conflate() drops intermediate values when the collector is slow, keeping only the most recent emission. Useful for UI state.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        repeat(5) { i -> emit(i); delay(50) }
    }.conflate()
    .collect { i ->
        delay(150)
        println("Collected: $i")  // skips some values
    }
}

collectLatest for Slow Collectors

collectLatest cancels the current collection block when a new value arrives, restarting with the latest value.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        emit(1); delay(50)
        emit(2); delay(50)
        emit(3)
    }.collectLatest { value ->
        println("Processing $value")
        delay(100)  // gets cancelled if new value arrives
        println("Done $value")  // only prints for last value
    }
}

flowOn + buffer Pattern

Combine flowOn and buffer: run the producer on IO (network/disk), buffer results, collect on Main.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun fetchItems(): Flow<String> = flow {
    repeat(3) { i ->
        delay(100)  // simulate IO
        emit("item-$i")
    }
}.flowOn(Dispatchers.IO).buffer(10)
fun main() = runBlocking {
    fetchItems().collect { println("UI: $it") }
}

channelFlow for Concurrent Producers

channelFlow creates a flow backed by a channel, allowing multiple coroutines to emit concurrently from within the builder.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun concurrentFlow(): Flow<Int> = channelFlow {
    launch { send(1) }
    launch { send(2) }
    launch { send(3) }
}
fun main() = runBlocking {
    concurrentFlow().collect { println(it) }
}

Choosing the Right Strategy

Summary: flowOn for context switching, buffer for throughput, conflate for latest-only UI, collectLatest for cancelling stale processing.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
// Guidelines:
// CPU-heavy production  -> flowOn(Dispatchers.Default)
// IO-heavy production   -> flowOn(Dispatchers.IO)
// Slow consumer         -> buffer()
// UI state updates      -> conflate() or StateFlow
// Search/autocomplete   -> collectLatest or flatMapLatest
fun main() = runBlocking { println("Choose the right strategy!") }

Quick Check

What does flowOn affect in a flow chain?

Answer it in the interactive lesson

Recap

flowOn moves upstream work to another dispatcher. buffer decouples producer and consumer for throughput. conflate drops intermediate values. collectLatest cancels slow processing on new arrivals.

Free to start

Learn Kotlin with an AI tutor — free

Write and run real code in your browser, get instant help from a 24/7 AI tutor, and pick up where you left off on the web or in the app.

Courses: 51
Lessons: 203

Start learning free View the full course

Frequently Asked Questions

Is the “flowOn and buffer: Context and Backpressure” lesson free?

Yes — any 3 lessons of the Kotlin Academy learning path, including “flowOn and buffer: Context and Backpressure”, are free to read in full here on the web. After that, CoddyKit PRO unlocks every lesson, plus interactive practice with a built-in code editor and a 24/7 AI tutor. The Kotlin Academy course includes 4 lessons in total.

What will I learn in “flowOn and buffer: Context and Backpressure”?

Change emission context with flowOn and buffer emissions for backpressure. You practise Kotlin Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start Kotlin Academy?

No prior experience is required. Kotlin Academy on CoddyKit is structured for beginners through advanced learners, so you can start here or from the beginning and move at your own pace. This is lesson 4 of 4.

How long does the “flowOn and buffer: Context and Backpressure” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this Kotlin Academy lesson?

Yes. Every Kotlin Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

Flow Operators: map, filter, transform, and take
catch and onCompletion: Error Handling in Flow
combine and zip: Merging Multiple Flows
flowOn and buffer: Context and Backpressure

← Back to Kotlin Academy