Kotlin Flow Explained
Flow is coroutines' streaming extension — cold streams emit on demand, backpressure is built in, and operators are composable. Charlie uses Flow to build a real-time order processing pipeline.
1. What You'll Learn
- Cold stream vs hot stream:
Flowemits on demand - Intermediate operations:
map/filter/transform/take/debounce - Terminal operations:
collect/toList/first/single - Backpressure handling:
buffer/conflate/collectLatest - Charlie in action: order event streaming pipeline
2. A Real Architect's Story
(1) Pain Point: Real-Time Order Processing Bottleneck
Charlie's OrderProcessor needs to process 5,000 order events per second in real time. Batch processing with List has high latency, callback nesting is unmaintainable, and RxJava has a steep learning curve.
(2) The Flow Solution
KOTLIN
// Flow: cold stream, on-demand emission, backpressure built-in
fun orderEventStream(): Flow<OrderEvent> = flow {
while (true) {
val event = pollOrderEvent() // Non-blocking poll
emit(event) // Emit to downstream
}
}
// Compose pipeline: filter -> transform -> collect
orderEventStream()
.filter { it.total > 100 }
.map { enrich(it) }
.collect { dispatch(it) }
Flow is coroutine-native streaming — cold emission on demand, built-in backpressure, composable operators, far lower learning curve than RxJava.
3. Flow Basics
(1) Creating Flows
KOTLIN
// 1. flow builder
fun numberFlow(): Flow<Int> = flow {
for (i in 1..5) {
emit(i) // Emit value to collector
delay(100) // Simulate work
}
}
// 2. flowOf: from fixed values
val orderFlow = flowOf(
Order("ORD-001", 299.99),
Order("ORD-002", 1_500.00)
)
// 3. asFlow: from Iterable/Sequence
val orderList = listOf(Order("ORD-001", 299.99))
val orderFlow2 = orderList.asFlow()
// 4. channelFlow: for complex concurrent emission
fun concurrentFlow(): Flow<Int> = channelFlow {
launch { send(1) }
launch { send(2) }
}
(2) Cold Stream Behavior
KOTLIN
// Cold flow: each collector triggers new emission
val coldFlow = flow {
println("Emitting...") // This runs EACH time collect is called
emit(1)
emit(2)
}
coldFlow.collect { println("Collector 1: $it") } // Emits 1, 2
coldFlow.collect { println("Collector 2: $it") } // Emits 1, 2 AGAIN
(3) Cold Flow vs Hot Flow
| Dimension | Cold Flow (Flow) | Hot Flow (SharedFlow/StateFlow) |
|---|---|---|
| Trigger | Emits only when collected | Runs independently of collectors |
| Data | Each collector gets full data | New collectors only see subsequent data |
| Caching | None | Configurable replay |
| Lifecycle | Follows collector | Independent |
| Analogy | Video on demand | Live stream |
4. Intermediate Operations
Intermediate operations are cold — they don't trigger emission, only define transformation logic.
(1) map / filter / transform
KOTLIN
// map: transform each value
val totals = orderFlow.map { it.total }
// filter: keep matching values
val highValue = orderFlow.filter { it.total > 1_000 }
// transform: more flexible than map (can emit multiple values)
val expanded = orderFlow.transform { order ->
emit("Order: ${order.id}")
emit("Total: \$${order.total} USD")
}
(2) take / drop / debounce
KOTLIN
// take: take first N values
val first3 = orderFlow.take(3)
// drop: skip first N values
val after5 = orderFlow.drop(5)
// debounce: emit only after silence period
val debounced = searchFlow.debounce(300) // Wait 300ms of no new values
(3) Intermediate Operation Categories
| Category | Operators | Function |
|---|---|---|
| Transform | map / transform |
Per-value transformation |
| Filter | filter / filterNot / filterIsInstance |
Conditional filtering |
| Flatten | flatMapConcat / flatMapMerge / flatMapLatest |
Nested stream flattening |
| Truncate | take / drop / takeWhile |
Take/skip first N |
| Timing | debounce / sample / throttleFirst |
Time-based control |
| Combine | zip / combine |
Multi-stream combination |
5. Terminal Operations
Terminal operations trigger actual emission and collection.
(1) collect / toList / first
KOTLIN
// collect: consume each value
orderFlow.collect { order -> println(order.id) }
// toList: collect to list
val allOrders = orderFlow.toList()
// first: get first value
val firstOrder = orderFlow.first()
// single: expect exactly one value
val onlyOrder = orderFlow.single()
(2) Terminal Operation Categories
| Operation | Return Type | Description |
|---|---|---|
collect |
Unit | Consume each value |
toList / toSet |
List / Set |
Collect to collection |
first / firstOrNull |
T / T? |
Get first value |
single / singleOrNull |
T / T? |
Get unique value |
count |
Int | Count elements |
fold / reduce |
R / T |
Accumulate |
6. Backpressure Handling
(1) The Backpressure Problem
KOTLIN
// Producer is faster than consumer
val fastFlow = flow {
repeat(10) {
emit(it) // Fast: 1ms per emit
delay(1)
}
}
// Slow consumer causes backpressure
fastFlow.collect {
delay(100) // Slow: 100ms per process
println("Processed: $it")
}
// By default: producer waits for consumer (suspend)
(2) Backpressure Strategies
KOTLIN
// buffer: run producer and consumer in parallel
fastFlow.buffer(10) // Buffer up to 10 values
.collect { delay(100); println("Buffered: $it") }
// conflate: skip intermediate values, keep latest
fastFlow.conflate()
.collect { delay(100); println("Latest: $it") }
// collectLatest: cancel previous collection when new value arrives
fastFlow.collectLatest { value ->
delay(100)
println("Latest processed: $value")
}
(3) Backpressure Strategy Comparison
| Strategy | Behavior | Data Loss | Use Case |
|---|---|---|---|
| Default | Producer waits for consumer | No loss | Guarantee completeness |
buffer |
Buffers N values | Pauses when buffer full | Short production spikes |
conflate |
Skip intermediate, keep latest | Intermediate values lost | Real-time display |
collectLatest |
Cancel old processing, handle latest | Old processing interrupted | Search suggestions |
7. Flow Cold Stream Pipeline
flowchart LR
A[Source<br/>flow emit] --> B[filter<br/>total > 100]
B --> C[map<br/>enrich]
C --> D[debounce<br/>300ms]
D --> E[collect<br/>dispatch]
8. Complete Example: OrderProcessor Event Stream
KOTLIN
// ============================================
// OrderProcessor - Event Stream with Flow
// Feature: Real-time order event processing
// ============================================
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
data class Order(val id: String, val total: Double, val status: String, val customer: String)
sealed class OrderEvent {
data class Created(val order: Order) : OrderEvent()
data class Updated(val orderId: String, val newStatus: String) : OrderEvent()
data class Cancelled(val orderId: String, val reason: String) : OrderEvent()
}
// Simulate event source
fun orderEventStream(): Flow<OrderEvent> = flow {
val events = listOf(
OrderEvent.Created(Order("ORD-001", 299.99, "PENDING", "Alice")),
OrderEvent.Created(Order("ORD-002", 15_000.00, "PENDING", "Bob")),
OrderEvent.Updated("ORD-001", "CONFIRMED"),
OrderEvent.Created(Order("ORD-003", 2_500.00, "PENDING", "Charlie")),
OrderEvent.Cancelled("ORD-003", "Out of stock"),
OrderEvent.Updated("ORD-002", "CONFIRMED"),
OrderEvent.Created(Order("ORD-004", 8_900.00, "PENDING", "Bob")),
OrderEvent.Updated("ORD-002", "SHIPPED")
)
events.forEach { event ->
emit(event)
delay(100)
}
}
// Enrich created orders
fun Order.enrich() = copy(status = "ENRICHED")
fun main() = runBlocking {
println("=== Order Event Stream ===")
// Pipeline: filter created -> enrich -> collect
orderEventStream()
.onStart { println("Stream started") }
.onEach { println(" Received: $it") }
.filter { it is OrderEvent.Created }
.map { (it as OrderEvent.Created).order }
.filter { it.total > 1_000 }
.map { it.enrich() }
.onEach { println(" >>> Enriched high-value: ${it.id} (\$${it.total} USD)") }
.onCompletion { println("Stream completed") }
.collect()
// Separate pipeline: status updates
println("\n=== Status Updates ===")
orderEventStream()
.filter { it is OrderEvent.Updated }
.map { it as OrderEvent.Updated }
.collect { event ->
println(" ${event.orderId} -> ${event.newStatus}")
}
// Aggregate: count events by type
println("\n=== Event Counts ===")
val counts = orderEventStream()
.groupBy { it::class.simpleName ?: "Unknown" }
.map { (type, flow) ->
type to flow.count()
}
.toList()
.associate { it }
counts.forEach { (type, count) ->
println(" $type: $count events")
}
}
Output:
TEXT
=== Order Event Stream ===
Stream started
Received: Created(order=Order(ORD-001, 299.99, PENDING, Alice))
Received: Created(order=Order(ORD-002, 15000.0, PENDING, Bob))
>>> Enriched high-value: ORD-002 ($15000.0 USD)
Received: Updated(orderId=ORD-001, newStatus=CONFIRMED)
Received: Created(order=Order(ORD-003, 2500.0, PENDING, Charlie))
>>> Enriched high-value: ORD-003 ($2500.0 USD)
Received: Cancelled(orderId=ORD-003, reason=Out of stock)
Received: Updated(orderId=ORD-002, newStatus=CONFIRMED)
Received: Created(order=Order(ORD-004, 8900.0, PENDING, Bob))
>>> Enriched high-value: ORD-004 ($8900.0 USD)
Received: Updated(orderId=ORD-002, newStatus=SHIPPED)
Stream completed
=== Status Updates ===
ORD-001 -> CONFIRMED
ORD-002 -> CONFIRMED
ORD-002 -> SHIPPED
=== Event Counts ===
Created: 4 events
Updated: 3 events
Cancelled: 1 events
❓ FAQ
Q What's the difference between Flow and Sequence?
A Flow is asynchronous (supports suspend), Sequence is synchronous. Flow suits async data sources (network/database); Sequence suits synchronous big-data processing.
Q What's the difference between Flow and RxJava's Observable?
A Flow is coroutine-native, cold-stream-first, with a gentle learning curve. RxJava is more mature but more complex. Choose Flow for new projects; coexist with RxJava in existing projects.
Q When does a cold flow become a hot flow?
A Use
shareIn to convert a cold Flow to SharedFlow, or stateIn for StateFlow. This lets multiple collectors share the same data source.Q How to choose between flatMapConcat, flatMapMerge, flatMapLatest?
A
flatMapConcat processes sequentially (default), flatMapMerge processes concurrently (higher throughput), flatMapLatest cancels old streams (real-time scenarios like search).Q How does Flow handle backpressure by default?
A Default behavior is "producer waits for consumer" —
emit is a suspend function; when the consumer is slow, the producer automatically suspends. This is the safest default.Q Is collect a suspend function?
A Yes.
collect is a suspend function, callable only from coroutines or suspend functions. It suspends until the Flow completes.📖 Summary
- Flow is a cold stream: each collector triggers independent emission, no data sharing
- Intermediate operations (map/filter/transform) are cold — don't trigger emission
- Terminal operations (collect/toList/first) trigger actual data flow
- Default backpressure: producer waits for consumer; adjust with buffer/conflate/collectLatest
- Flow operator syntax is consistent with collection operators — low learning curve
onStart/onEach/onCompletionare stream lifecycle hooks
📝 Exercises
- Beginner (⭐): Create a Flow that emits 1-10 using
flow { }, then filter and print even numbers. Hint:flow { for (i in 1..10) emit(i) }.filter { } - Intermediate (⭐⭐): Simulate an order event stream with Flow, use
filter+map+collectto process orders with amounts > 1000. Hint: refer to Section 8 - Advanced (⭐⭐⭐): Implement a search stream with
debounce: simulate user input, trigger search only after 300ms of no new input. Hint:debounce(300)
