Kotlin 協程 (三)

協程通信是協程之間進行數據交換和同步的關鍵機制。Kotlin 協程提供了多種通信方式，使得協程能夠高效、安全地進行交互。以下是對協程通信的詳細講解，包括常見的通信原語、使用場景和示例代碼。

1.1 Channel

定義：Channel 是一個消息隊列，用于協程之間的通信。它允許協程發送和接收數據，類似于 MPI（Message Passing Interface）中的消息傳遞機制。

特點：

線程安全。
支持多種通信模式，如無緩沖、有緩沖和無限緩沖。
提供掛起函數?send?和?receive，用于發送和接收數據。
可以關閉。

Channel 類型

類型	描述	創建方式	適用場景
Rendezvous	無緩沖（默認）	`Channel<T>()`	嚴格的發送-接收同步
Buffered	固定大小緩沖	`Channel<T>(capacity)`	控制內存使用
Conflated	保留最新值	`Channel<T>(CONFLATED)`	只需要最新數據
Unlimited	無限緩沖	`Channel<T>(UNLIMITED)`	生產者快于消費者
Broadcast	廣播給多個接收者	`BroadcastChannel<T>(capacity)`	一對多通信

使用場景：

1.生產者-消費者模式

場景描述：一個或多個生產者協程生成數據，一個或多個消費者協程處理數據。
適用性：當數據需要按順序處理時，Channel?提供了天然的 FIFO 隊列。
示例：文件處理流水線，其中文件讀取、處理和寫入由不同的協程完成。

2 .協程間的事件總線

場景描述：一個協程發送事件，多個協程監聽并響應這些事件。
適用性：當需要解耦協程之間的通信時，Channel?可以作為事件傳遞的媒介。
示例：在 GUI 應用中，用戶操作（如按鈕點擊）通過?Channel?發送事件，不同的協程根據事件執行相應的邏輯。

3. 資源池管理

場景描述：多個協程需要共享一組有限資源，如數據庫連接或網絡請求。
適用性：Channel?可以控制對資源的并發訪問，避免資源爭用導致的沖突。
示例：使用?Channel?作為資源池，協程從池中請求資源，使用完畢后釋放回池中

示例：

fun main() = runBlocking {val channel = Channel<Int>() // 創建無緩沖通道// 生產者協程launch {for (x in 1..5) {println("Sending $x")channel.send(x) // 掛起直到有接收者delay(100) // 模擬工作}channel.close() // 關閉通道}// 消費者協程launch {for (y in channel) { // 使用for循環接收println("Received $y")}println("Channel is closed")}delay(2000)
}

1.2 Flow

定義：Flow?是一種冷流（cold stream），意味著數據流只有在收到收集（collection）請求時才會開始發射數據。它支持掛起操作，可以與協程完美結合，實現異步計算和數據處理。

特點：

異步數據流：以異步方式處理連續數據流。
聲明式編程：通過操作符鏈式調用處理數據流。
可組合性：支持 map、filter、flatMap、zip 等操作符。
取消支持：與協程一樣支持取消操作。

基本使用：

//使用 flow 建造器創建一個 Flow 對象。
fun transformFlow() = flow {for (i in 1..5) {delay(100)emit(i)}
}.map { it * 2 } // 將每個值乘以 2.filter { it > 4 } // 過濾掉小于等于 4 的值fun main() = runBlocking {
//使用 collect 函數收集 Flow 發射的數據。transformFlow().collect { value -> println("Transformed value: $value")}
}

使用場景：

1 .異步數據流處理

場景描述：對異步數據流進行轉換、合并、過濾等操作。
適用性：當需要對數據流進行復雜的操作時，Flow?提供了豐富的操作符。
示例：在數據處理管道中，使用?Flow?對數據進行映射、過濾和歸約。

2. 網絡請求與數據解析

場景描述：從網絡獲取數據，并進行解析和轉換。
適用性：Flow?可以與網絡庫（如 Retrofit）結合，簡化異步網絡請求的處理。
示例：使用?Flow?處理分頁網絡請求，逐頁獲取數據并進行解析。

3.?數據庫查詢與觀察

場景描述：從數據庫查詢數據，并對結果進行觀察。
適用性：Flow?可以與數據庫庫（如 Room）結合，實現數據的異步查詢和實時更新。
示例：在 Android 應用中，使用?Flow?監聽數據庫表的變化，并更新 UI。

示例：

import retrofit2.Retrofit
import retrofit2.converter.gson.GsonConverterFactory
import retrofit2.http.GET
import kotlinx.coroutines.*interface ApiService {@GET("data")fun fetchData(): Flow<String>
}val apiService = Retrofit.Builder().baseUrl("https://api.example.com").addConverterFactory(GsonConverterFactory.create()).build().create(ApiService::class.java)fun main() = runBlocking {apiService.fetchData().flowOn(Dispatchers.IO) // 在 IO 線程上執行網絡請求.collect { data ->println("Received data: $data")}
}

1.3 SharedFlow 和 StateFlow

定義：SharedFlow 和 StateFlow 是 Kotlin Flow 庫中的兩種特殊 Flow，用于在多個收集器之間共享狀態和數據流。

特點：

SharedFlow：熱流，允許多個收集器接收數據，可配置重放（replay）和緩沖，不保存狀態。
StateFlow：特殊的 SharedFlow，必須有初始值，只保留最新值，并在收集器加入時立即發出當前狀態。

使用場景：

1.?事件流共享

場景描述：多個協程需要共享同一個事件流，并且每個協程都能接收到事件。
適用性：當事件需要被多個訂閱者處理，且每個訂閱者都能獨立地接收事件時。
示例：在實時數據應用中，傳感器數據通過?SharedFlow?分發給多個處理單元。

2?狀態廣播

場景描述：一個協程更新狀態，多個協程監聽狀態變化。
適用性：當狀態變化需要通知給多個觀察者時，SharedFlow?提供了廣播機制。
示例：在游戲開發中，玩家狀態（如生命值、分數）通過?SharedFlow?廣播給 UI 和邏輯處理單元。

3. 背壓管理

場景描述：生產者生成數據的速度可能快于消費者處理數據的速度。
適用性：SharedFlow?支持背壓策略，可以控制數據的發送速度，避免消費者過載。
示例：在數據流處理中，使用?SharedFlow?緩解高速數據源對處理單元的壓力。

示例：

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*fun main() = runBlocking {val sharedFlow = MutableSharedFlow<String>()// 事件發送協程launch {sharedFlow.emit("Event 1")delay(1000)sharedFlow.emit("Event 2")}// 事件監聽協程 1launch {sharedFlow.collect { event ->println("Listener 1 received: $event")}}// 事件監聽協程 2launch {sharedFlow.collect { event ->println("Listener 2 received: $event")}}delay(2000) // 等待事件處理完成coroutineContext.cancelChildren() // 取消所有子協程
}

1.4?Await 和 Async

定義：async?是一種啟動協程并獲取其結果的方式。await?用于等待?async?協程的結果。

特點：

async?會立即返回一個?Deferred?對象，可以在需要時通過?await?獲取結果。
支持結構化并發，await?會在需要時掛起協程，直到結果準備好。

使用場景：

當需要并行執行任務并聚合結果時。
當需要按需獲取協程結果時。

示例：

 import kotlinx.coroutines.*suspend fun main() {val deferred = async { expensiveComputation() }val result = deferred.await()println("Result: $result")}suspend fun expensiveComputation(): Int {delay(1000)return 42}

1.5 Actor

定義：Actor 是結合了協程和通道的實體，封裝了狀態和處理消息的能力。

特點：

消息傳遞：Actor 之間通過發送和接收消息進行通信。
封裝狀態：每個 Actor 封裝了自己的狀態和行為，其他 Actor 無法直接訪問其內部狀態。
異步通信：通過消息傳遞實現異步交互，避免傳統并發模型中的死鎖和競爭問題。

使用場景：

分布式系統：在分布式環境中，Actor 可以作為獨立的計算單元，通過消息傳遞完成協同任務。
實時通信：適用于需要實時處理消息的場景，如聊天應用、游戲服務器等。
高并發任務處理：處理需要同時執行多個任務的應用，例如訂單處理、數據分析等。

示例：

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*fun main() = runBlocking {val actor = actor<String> {//Actor 內部通過 for 循環接收來自 channel 的消息，并執行相應的處理邏輯。for (msg in channel) {println("Received: $msg")// 處理消息}}// 向 Actor 發送消息actor.send("Hello")actor.send("World")actor.close() // 關閉 Actor
}

2. 示例：實時搜索場景

class SearchViewModel : ViewModel() {// 使用 MutableStateFlow 存儲可變的搜索查詢字符串// StateFlow 是熱流，會自動收集數據，適合 UI 狀態管理private val searchQuery = MutableStateFlow("")// 使用 MutableStateFlow 存儲可變的搜索結果列表// 注意：這里返回的 searchResults 是不可變的 StateFlow，避免外部直接修改private val _searchResults = MutableStateFlow<List<String>>(emptyList())val searchResults: StateFlow<List<String>> = _searchResultsinit {// 在 viewModelScope 中啟動協程// viewModelScope 是 ViewModel 提供的協程作用域，基于 Dispatchers.Main 運行// 當 ViewModel 銷毀時，viewModelScope 會自動取消所有協程，避免內存泄漏viewModelScope.launch {// 構建 Flow 管道，處理搜索查詢到結果的轉換searchQuery// debounce(300)：防抖處理，300ms 內多次輸入只觸發最后一次// 避免用戶快速輸入時觸發過多搜索請求.debounce(300)// filter it.length > 2：過濾短查詢，避免無效請求// 只有查詢長度大于 2 時才繼續處理.filter { it.length > 2 }// distinctUntilChanged()：忽略重復的查詢，避免重復請求// 只有查詢內容變化時才繼續處理.distinctUntilChanged()// flatMapLatest：取消前一個未完成的搜索，只保留最新的查詢結果// 當新的查詢到來時，會取消前一個協程任務.flatMapLatest { query ->// 調用 performSearch 發起搜索，返回 Flow<List<String>>performSearch(query)}// catch emit(emptyList())：捕獲異常并返回空列表，避免 UI 出錯// 如果搜索過程中發生異常，會發射空列表作為默認結果.catch { emit(emptyList()) }// collect：收集 Flow 發射的數據，更新搜索結果.collect { results ->// 更新 _searchResults 的值，UI 會自動響應變化_searchResults.value = results}}}// 處理用戶輸入變化，更新搜索查詢fun onSearchQueryChanged(query: String) {// 直接設置 MutableStateFlow 的值，會觸發 Flow 管道重新計算searchQuery.value = query}// 模擬網絡搜索請求，返回 Flow<List<String>>private fun performSearch(query: String): Flow<List<String>> = flow {// delay(1000)：模擬耗時操作（如網絡請求），可被協程取消// 如果協程被取消，delay 會立即拋出 CancellationExceptiondelay(1000)// emit：發射搜索結果// 這里模擬返回兩個結果項emit(listOf("query result 1", "query result 2"))}
}