Chapter 10 : UseCase07 - Buffer and Backpressure
Concepts to Learn
What backpressure is
Why slow collectors block fast emitters
How buffer() changes behaviour
How producer and consumer can run independently
Let us first write the simplest code for flow and collect that we have seen so far.
package org.kotlinflowlearner.stockflow.usecases.uc07
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
class UC07Buffer {
fun simpleFlow(): Flow<Int> = flow {
repeat(10) {
println("Emitting $it")
emit(it)
}
}
}
package org.kotlinflowlearner.stockflow.usecases.uc07
import kotlinx.coroutines.runBlocking
fun main(){
runBlocking {
val useCase = UC07Buffer()
println("---- Without Buffer And Delay ----")
useCase.simpleFlow().collect{
value -> println("Collected $value")
}
}
}
In a normal Flow without buffer(), the producer and the consumer move one-by-one together. In the above example we have a loop that operates 10 times, Every sequential emission (producer) starting 0 is followed by a collection (consumer) of the same integer. In this ideal scenario, the producer-consumer work without buffer and zero delays.
The output is something like this:
--- Without Buffer And Delay ----
Emitting 0
Collected 0
Emitting 1
Collected 1
Emitting 2
Collected 2
Emitting 3
Collected 3
Emitting 4
Collected 4
Emitting 5
Collected 5
Emitting 6
Collected 6
Emitting 7
Collected 7
Emitting 8
Collected 8
Emitting 9
Collected 9
That means:
The emitter produces one value.
It waits.
The collector processes it.
Only then can the emitter produce the next value.
Neither side can move independently. In the above case the wait time is zero. In a real world scenario, there will be a delay.
In a real-world problem, the emitter produces one value, and then it must wait until the collector finishes processing that value before emitting the next one. This means the whole pipeline behaves sequentially.
Now let us add another function without buffer but has a delay. This mimics a real-world problem where there are delays.
package org.kotlinflowlearner.stockflow.usecases.uc07
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
class UC07Buffer {
fun simpleFLow(): Flow<Int> = flow {
repeat(10) {
println("Emitting $it")
emit(it)
}
}
fun withoutBufferButDelay(): Flow<Int> = flow {
repeat(5) {
println("Emitting $it")
emit(it)
delay(100) // Fast producer
}
}
}
package org.kotlinflowlearner.stockflow.usecases.uc07
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
fun main(){
runBlocking {
val useCase = UC07Buffer()
println("---- Without Buffer And Delay ----")
useCase.simpleFLow().collect{
value -> println("Collected $value")
}
println("---- Without Buffer But Delay ----")
useCase.withoutBufferButDelay().collect{
value ->
delay(300) // Slow consumer
println("Collected $value")
}
}
}
In the function withoutBufferButDelay() , the producer emits a number every 100 milliseconds. However, the collector waits 300 milliseconds before printing the value. Because there is no buffer, the emitter cannot run ahead. It emits 0 and then suspends while the collector spends 300 milliseconds processing it. Only after the collector resumes does the emitter continue with 1. So even though the producer is “fast,” it is forced to behave slowly. The slowest component controls the total speed of the flow.
What Happens Without Buffer
Sequence becomes:
Emit 0
Wait 300ms (collector slow)
Emit 1
Wait 300ms
Emit 2
Producer emits every 100ms. Collector processes every 300ms. Producer is forced to wait.
Total time is approximately equal to (producer delay + consumer delay combined).
Everything is sequential.
The slow collector is pushing back on the fast producer. That “push back” effect is called backpressure.
When the collector suspends, upstream is also suspended. So the consumer controls the pace. The slowest stage determines the total throughput. That is backpressure in action.
package org.kotlinflowlearner.stockflow.usecases.uc07
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.flow
class UC07Buffer {
fun simpleFLow(): Flow<Int> = flow {
repeat(10) {
println("Emitting $it")
emit(it)
}
}
fun withoutBufferButDelay(): Flow<Int> = flow {
repeat(5) {
println("Emitting $it")
emit(it)
delay(100) // Fast producer
}
}
fun withBuffer() : Flow<Int> = flow {
repeat(5){
println("Emiting $it")
emit(it)
delay(100)
}
}.buffer()
}
package org.kotlinflowlearner.stockflow.usecases.uc07
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
fun main(){
runBlocking {
val useCase = UC07Buffer()
println("---- Without Buffer And Delay ----")
useCase.simpleFLow().collect{
value -> println("Collected $value")
}
println("---- Without Buffer But Delay ----")
useCase.withoutBufferButDelay().collect{
value ->
delay(300) // Slow consumer
println("Collected $value")
}
println("---- With Buffer ----")
useCase.withBuffer().collect{
value -> delay(300)
println("Collected $value")
}
}
}
What Happens With Buffer
Producer emits rapidly:
Emit 0
Emit 1
Emit 2
Emit 3
Emit 4
While consumer processes slowly. Producer and consumer now overlap.
Total execution time decreases.
Improvements made
Without buffer : Producer and consumer are tightly coupled.
With buffer : They are decoupled. Buffer creates a small queue between the producer and consumer.
When we introduce buffer(), we insert a small queue between the emitter and the collector. Now, when the producer emits a value, it can place that value into the buffer and continue emitting the next one without waiting immediately for the collector. Meanwhile, the collector consumes values from the buffer at its own pace. The producer and consumer are no longer tightly coupled.
This does not mean the producer becomes infinitely fast. The buffer has limited capacity. If the buffer fills up because the consumer is too slow, the producer will eventually suspend again. However, as long as there is space in the buffer, upstream and downstream can operate concurrently.
The key idea is that Flow is sequential by default. Each emission waits for downstream processing to complete. Adding buffer() introduces controlled concurrency by allowing emissions to be temporarily stored instead of immediately processed. This is how Kotlin Flow handles backpressure in a structured way.
We can conclude that Backpressure is absorbed temporarily by the buffer.
PRACTICAL APPLICATION
In practical terms, imagine stock prices updating rapidly while an analytics engine processes them slowly. Without buffering, every update would wait for analytics to finish. With buffering, updates can accumulate briefly while processing continues independently. That is the real purpose of buffer().
This use case teaches us that performance in reactive systems is not only about speed, but also about coordination between producers and consumers.
SUMMARY
Flow is sequential by default.
Buffer introduces concurrency.
Buffer does not create infinite storage, it has capacity limits.
If capacity fills, suspension still happens.
Producer-consumer decoupling happens because of buffer()
Over to Chapter 11 now !!