一、 Flow 與 Channel 的相互轉換
1.1 Flow 轉換為 Channel
1.1.1 ChannelFlow
@InternalCoroutinesApi
public abstract class ChannelFlow<T>(
// upstream context
@JvmField public val context: CoroutineContext,
// buffer capacity between upstream and downstream context
@JvmField public val capacity: Int,
// buffer overflow strategy
@JvmField public val onBufferOverflow: BufferOverflow
) : FusibleFlow<T> {
...
public open fun produceImpl(scope: CoroutineScope): ReceiveChannel<T> =
scope.produce(context, produceCapacity, onBufferOverflow, start = CoroutineStart.ATOMIC, block = collectToFun)
...
}
前面提到 ChannelFlow 是熱流。只要上游產生數據，就會立即發射給下游收集者。
ChannelFlow 是一個抽象類，並且被標記為內部 Api，不應該在外部代碼直接使用。
注意到它內部有一個方法 produceImpl 返回的是一個 ReceiveChannel，它的實現是收集上游發射的數據，然后發送到 Channel 中。
有此作為基礎。我們可以 調用 asChannelFlow 將 Flow 轉換 ChannelFlow， 進而轉換成 Channel 。
1.1.2 produceIn —— 將 Flow 轉換為單播式 Channel
produceIn()轉換創建了一個produce 協程來 collect 原Flow，因此該produce協程應該在恰當時候被關閉或者取消。轉換后的 Channel 擁有處理背壓的能力。其基本使用方式如下：
fun main() = runBlocking {
val flow = flow<Int> {
repeat(5) {
delay(500)
emit(it)
}
}
val produceIn = flow.produceIn(this)
for (ele in produceIn) {
println(ele)
}
}
輸出結果：
0
1
2
3
4
查看 produceIn 源碼：
@FlowPreview
public fun <T> Flow<T>.produceIn(scope: CoroutineScope): ReceiveChannel<T> = asChannelFlow().produceImpl(scope)
1.1.3 broadcastIn —— 將 Flow 轉換為廣播式 BroadcastChannel。
broadcastIn 轉換方式與 produceIn 轉換方式實現原理一樣，區別是創建出來的 BroadcastChannel。
源碼如下：
public fun <T> Flow<T>.broadcastIn(
scope: CoroutineScope,
start: CoroutineStart = CoroutineStart.LAZY
): BroadcastChannel<T> {
// Backwards compatibility with operator fusing
val channelFlow = asChannelFlow()
val capacity = when (channelFlow.onBufferOverflow) {
BufferOverflow.SUSPEND -> channelFlow.produceCapacity
BufferOverflow.DROP_OLDEST -> Channel.CONFLATED
BufferOverflow.DROP_LATEST ->
throw IllegalArgumentException("Broadcast channel does not support BufferOverflow.DROP_LATEST")
}
return scope.broadcast(channelFlow.context, capacity = capacity, start = start) {
collect { value ->
send(value)
}
}
}
使用方式見上文 BroadcastChannel。
和 BroadcastChannel 一樣，broadcastIn 也標記為過時的 API, 不建議繼續使用了。
1.2 Channel 轉換為 Flow
1.2.1 consumeAsFlow/receiveAsFlow —— 將單播式 Channel 轉換為 Flow
使用 consumeAsFlow()/receiveAsFlow() 將 Channel 轉換為 Flow
fun main() = runBlocking<Unit> {
val testChannel = Channel<String>()
val testFlow = testChannel.receiveAsFlow()
launch {
testFlow.collect {
println(it)
}
}
delay(100)
testChannel.send("hello")
delay(100)
testChannel.send("coroutine")
delay(100)
testChannel.close() // 注意只有 Channel 關閉了，協程才能結束
}
查看源碼：
public fun <T> ReceiveChannel<T>.consumeAsFlow(): Flow<T> = ChannelAsFlow(this, consume = true)
public fun <T> ReceiveChannel<T>.receiveAsFlow(): Flow<T> = ChannelAsFlow(this, consume = false)
private class ChannelAsFlow<T>(
private val channel: ReceiveChannel<T>,
private val consume: Boolean,
context: CoroutineContext = EmptyCoroutineContext,
capacity: Int = Channel.OPTIONAL_CHANNEL,
onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
) : ChannelFlow<T>(context, capacity, onBufferOverflow) {}
consumeAsFlow 和 receiveAsFlow 都是調用 ChannelAsFlow 將 Channel 轉換成了 ChannelFlow，所以轉換結果是熱流。但它們傳遞的第二個參數 consume 不一樣。兩者區別如下：
- 使用
consumeAsFlow()轉換成的 Flow 只能有一個收集器收集，如果有多個收集器收集，將會拋出如下異常：
Exception in thread "main" java.lang.IllegalStateException: ReceiveChannel.consumeAsFlow can be collected just once
- 使用
receiveAsFlow()轉換成的 Flow 可以有多個收集器收集，但是保證每個元素只能被一個收集器收集到，即單播式。
通俗點說，就是使用 consumeAsFlow() 只能有一個消費者。 使用 receiveAsFlow() 可以有多個消費者，但當向 Channel 中發射一個數據之后，收到該元素的消費者是不確定的。
1.2.2 asFlow —— 將廣播式 BroadcastChannel 轉換為 Flow
與單播式相對的就是廣播式，讓每個消費者都收到該元素，這就需要一個廣播式的 Chanel：BroadcastChanel。
BroadcastChannel 調用 asFlow() 方法即可將其轉換為 Flow。
由於該方法也被標記為過時了，替代方案有 SharedFlow 和 StateFlow。
二、SharedIn —— 將冷數據流轉換為熱數據流
將 flow 轉換為 SharedFlow,可以使用 SharedIn 方法：
public fun <T> Flow<T>.shareIn(
scope: CoroutineScope,
started: SharingStarted,
replay: Int = 0
): SharedFlow<T> {
...
}
參數解釋：
- CoroutineScope 用於共享數據流的 CoroutineScope。此作用域函數的生命周期應長於任何使用方，以使共享數據流在足夠長的時間內保持活躍狀態
- replay 每個新收集器的數據項數量
- started “啟動” 方式
啟動方式有：
public fun interface SharingStarted {
public companion object {
// 立即啟動，並且永遠不會自動停止
- public val Eagerly: SharingStarted = StartedEagerly()
// 第一個訂閱者注冊后啟動，並且永遠不會自動停止
- public val Lazily: SharingStarted = StartedLazily()
// 第一個訂閱者注冊后啟動，最后一個訂閱者取消注冊后停止
- public fun WhileSubscribed(
stopTimeoutMillis: Long = 0,
replayExpirationMillis: Long = Long.MAX_VALUE
): SharingStarted =
StartedWhileSubscribed(stopTimeoutMillis, replayExpirationMillis)
}
}
三、callbackFlow —— 將基於回調的 API 轉換為數據流
Kotlin 協程和 Flow 可以完美解決異步調用、線程切換的問題。設計接口時，可以類似 Rxjava 那樣，避免使用回調。比如 Room 在內的很多庫已經支持將協程用於數據流操作。對於那些還不支持的庫，也可以將任何基於回調的 API 轉換為協程。
callbackFlow 是一個數據流構建器，可以將基於回調的 API 轉換為數據流。
3.1 callbackFlow 的使用
舉例：
interface Result<T> {
fun onSuccess(t: T)
fun onFail(msg: String)
}
fun getApi(res: Result<String>) {
thread{
printWithThreadInfo("getApiSync")
Thread.sleep(1000) // 模擬耗時任務
res.onSuccess("hello")
}.start()
}
getApi() 是一個基於回調設計的接口。如何使用 callbackFlow 轉換為 Flow 呢？
fun getApi(): Flow<String> = callbackFlow {
val res = object: Result<String> {
override fun onSuccess(t: String) {
trySend(t)
close(Exception("completion"))
}
override fun onFail(msg: String) {
}
}
getApi(res)
// 一定要調用骨氣函數 awaitClose， 保證流一直運行。在`awaitClose` 中移除 API 訂閱，防止任務泄漏。
awaitClose {
println("close")
}
}
// 新的 Api 使用方式
fun main() = runBlocking<Unit> {
getApi().flowOn(Dispatchers.IO)
.catch {
println("getApi fail, cause: ${it.message}")
}.onCompletion {
println("onCompletion")
}.collect {
printWithThreadInfo("getApi success, result: $it")
}
}
這時候你可能有疑問了，這在流的內部不還是使用了基於接口的調用嗎，分明沒有更方便。看下面的例子，就能體會到了。
3.2 callbackFlow 實戰
Android 開發中有一個常見的場景：輸入關鍵字進行查詢。比如有個 EditText，輸入文字后，基於輸入的文字進行網絡請求或者數據庫查詢。
假設查詢數據的接口：
fun <T>query(keyWord: String): Flow<T> {
return flow {
//...
}
}
首先定義一個方法將 EditText 內容變化的回調轉換成 Flow
fun textChangeFlow(editText: EditText): Flow<String> = callbackFlow {
val watcher = object : TextWatcher {
override fun beforeTextChanged(s: CharSequence?, start: Int, count: Int, after: Int) {
}
override fun onTextChanged(s: CharSequence?, start: Int, before: Int, count: Int) {
}
override fun afterTextChanged(s: Editable?) {
s?.let {
trySend(s.toString())
}
}
}
editText.addTextChangedListener(watcher)
awaitClose {
editText.removeTextChangedListener(watcher)
}
}
使用：
scope.launch{
textChangeFlow(editText)
.debounce(300) // 防抖處理
.flatMapLatest { keyWord -> // 只對最新的值進行搜索
flow {
<String>query(keyWord)
}
}.collect {
// ... 處理最終結果
}
}
在這個過程中，我們可以充分使用 Flow 的各種變換，對我們的中間過程進行處理。實現一些很難實現的需求。