Mastering in Kotlin — Part-5- Sharing The data between coroutines

Sharing data between coroutines is a common requirement in concurrent programming, and Kotlin provides several approaches to achieve this. Each approach has its own advantages, disadvantages, and best use cases. Let’s explore the various methods of sharing data between coroutines and when to use each approach.

Share Data between Coroutines

• Shared Mutable State
• Actors
• Reactive Stream
• Channel

1. Shared Mutable State

Description:

Shared mutable state involves sharing data using shared variables like lists, maps, or other mutable data structures. However, you need to ensure proper synchronization to avoid race conditions and data corruption.

import kotlinx.coroutines.*

val sharedList = mutableListOf<Int>()

fun main() = runBlocking {
    val job1 = launch {
        repeat(10) {
            sharedList.add(it)
            delay(100)
        }
    }
    val job2 = launch {
        repeat(10) {
            sharedList.add(it + 10)
            delay(100)
        }
    }
    job1.join()
    job2.join()
    println(sharedList)
}

When to Use:

• Suitable for simple scenarios with low contention and small data sets.
• Avoid in complex scenarios with high contention or when multiple coroutines modify the same data concurrently.

2. Actors

Description:

Actors are a higher-level concurrency primitive that encapsulates state and behavior. They provide a structured way to share mutable state between coroutines while ensuring thread safety.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.actor

sealed class Message
data class Increment(val value: Int) : Message()
data class GetTotal(val response: CompletableDeferred<Int>) : Message()

fun CoroutineScope.counterActor() = actor<Message> {
    var total = 0
    for (msg in channel) {
        when (msg) {
            is Increment -> total += msg.value
            is GetTotal -> msg.response.complete(total)
        }
    }
}

fun main() = runBlocking {
    val counter = counterActor()
    repeat(5) {
        counter.send(Increment(1))
    }
    val response = CompletableDeferred<Int>()
    counter.send(GetTotal(response))
    println("Total: ${response.await()}")
    counter.close()
}

When to Use:

• Ideal for scenarios where you need to encapsulate state and behavior within a coroutine.
• Provides built-in message passing and state isolation, making it easier to reason about concurrent code.

3. Reactive Streams (e.g., Kotlin Flows)

Description:

Reactive streams libraries like Kotlin Flows provide asynchronous data streams and allow for declarative data processing and composition of asynchronous operations.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.measureTimeMillis

fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
    (1..5).asFlow()
        .map { requestFlow(it) }
        .collect { response ->
            println("Response: $response")
        }
    val endTime = System.currentTimeMillis()
    println("Total time: ${endTime - startTime} ms")
}

suspend fun requestFlow(i: Int): String {
    delay(1000) // Simulating network request
    return "Response $i"
}

When to Use:

• Suitable for asynchronous stream processing and reactive programming.
• Provides operators for transforming, filtering, and combining data streams in a declarative manner.

4. Channels

Description:

Channels provide a way for coroutines to communicate with each other in a non-blocking, thread-safe manner. They allow both sending and receiving of data between coroutines.

a. Creating a Channel

You can create a channel using the Channel factory function, specifying the type of elements that the channel will transmit.

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val channel = Channel<Int>()
    val job = launch {
        for (i in 1..5) {
            channel.send(i)
        }
        channel.close()
    }
    for (element in channel) {
        println(element)
    }
    job.join()
}

In this example, a channel of integers is created, and values 1 through 5 are sent through the channel. The channel is then closed, and the values are received and printed.

b. Buffered Channels

You can create buffered channels to store a limited number of elements in the channel buffer. This can help in scenarios where the sender and receiver may operate at different speeds.

val channel = Channel<Int>(capacity = 5) // Buffer size of 5

c. Channel Operations

Channels support various operations like send, receive, close, and offer. These operations allow you to send and receive elements, close the channel, and offer elements without blocking.

When to Use:

• Suitable for producer-consumer scenarios or when you need to stream data between coroutines.
• Provides flow control mechanisms like buffering and backpressure handling.

Choosing the Right Approach

• Shared Mutable State: Use for simple scenarios with low contention and small data sets. Avoid in complex scenarios with high contention.
• Channels: Ideal for producer-consumer scenarios or streaming data between coroutines.
• Actors: Use when you need to encapsulate state and behavior within a coroutine and ensure thread safety.
• Reactive Streams: Suitable for asynchronous stream processing and reactive programming, especially in UI or network-related tasks.