当前位置：首页 > news >正文

kotlin中SharedFlow的简单使用

news 2025/11/14 6:04:40

事件总线配置示例

对于一次性事件通知（如Toast消息、导航指令），推荐使用无replay缓存且支持背压处理的配置

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// 事件总线配置：不缓存历史事件，缓冲区满时丢弃最旧事件
object AppEventBus {
private val _events = MutableSharedFlow<AppEvent>(
replay = 0, // 新订阅者不接收历史事件
extraBufferCapacity = 5, // 额外缓冲区容量
onBufferOverflow = BufferOverflow.DROP_OLDEST // 确保最新事件及时处理
)

val events: SharedFlow<AppEvent> = _events.asSharedFlow()

suspend fun postEvent(event: AppEvent) {
_events.emit(event)
}
}

// 应用事件类型
sealed class AppEvent {
data class ShowToast(val message: String) : AppEvent()
data class NavigationEvent(val destination: String) : AppEvent()
data class PermissionResult(val permission: String, val granted: Boolean) : AppEvent()
}

// 使用示例
class MainActivity {
fun setupEventListeners() {
lifecycleScope.launch {
repeatOnLifecycle(Lifecycle.State.STARTED) {
AppEventBus.events.collect { event ->
when (event) {
is AppEvent.ShowToast -> showToast(event.message)
is AppEvent.NavigationEvent -> navigateTo(event.destination)
is AppEvent.PermissionResult -> handlePermissionResult(event)
}
}
}
}
}

private fun triggerEvent() {
lifecycleScope.launch {
AppEventBus.postEvent(AppEvent.ShowToast("操作成功"))
AppEventBus.postEvent(AppEvent.NavigationEvent("home"))
AppEventBus.postEvent(AppEvent.PermissionResult("CAMERA", true))
}
}

此配置适用于一次性事件分发，确保新订阅者只接收订阅后的事件，同时通过DROP_OLDEST策略保证高优先级事件的处理。

UI状态管理配置

对于需要保持最新状态的UI组件（如页面加载状态、表单数据），推荐使用带replay缓存的配置。

import kotlinx.coroutines.flow.*

class UserProfileViewModel : ViewModel() {
// UI状态配置：缓存最新状态，确保新订阅者立即获得当前状态
private val _uiState = MutableSharedFlow<UserState>(
replay = 1, // 为新订阅者提供最新状态
extraBufferCapacity = 3, // 额外状态缓冲区
onBufferOverflow = BufferOverflow.SUSPEND // 状态必须完整传递
)

val uiState: SharedFlow<UserState> = _uiState.asSharedFlow()

// 用户状态数据类
data class UserState(
val isLoading: Boolean = false,
val userData: User? = null,
val errorMessage: String? = null
)

fun updateUserProfile() {
viewModelScope.launch {
// 发送加载状态
_uiState.emit(UserState(isLoading = true))

// 模拟网络请求
delay(1000)

// 发送成功状态
_uiState.emit(UserState(
isLoading = false,
userData = User(name = "张三", email = "zhangsan@example.com")
))
}
}
}

这种配置保证新进入的UI组件能立即获得当前应用状态，避免状态不一致问题。

高频率数据流配置

对于传感器数据、实时日志等高频数据流，需采用优化的背压处理策略：

import kotlinx.coroutines.flow.*

class SensorDataProcessor {
// 高频数据流配置：快速处理最新数据，避免积压
private val _sensorFlow = MutableSharedFlow<SensorData>(
replay = 0, // 不缓存历史数据
extraBufferCapacity = 10, // 较大的缓冲区
onBufferOverflow = BufferOverflow.DROP_LATEST // 丢弃最新数据，保留处理中的序列
)

val sensorFlow: SharedFlow<SensorData> = _sensorFlow.asSharedFlow()

// 模拟高频数据产生
fun startProducingData() {
viewModelScope.launch {
var count = 0
while (true) {
_sensorFlow.emit(SensorData(
value = Math.random(),
timestamp = System.currentTimeMillis()
))
count++
delay(10) // 每10毫秒产生一个数据点
}
}
}
}

// 传感器数据结构
data class SensorData(
val value: Double,
val timestamp: Long
)

DROP_LATEST策略适用于消费者处理速度跟不上生产速度的场景，确保系统稳定性。

多订阅者广播配置

当需要向多个组件广播相同数据时（如埋点上报、全局配置更新），可采用以下配置：

import kotlinx.coroutines.flow.*

class GlobalConfigManager {
// 多订阅者配置：缓存配置历史，支持批量订阅
private val _configFlow = MutableSharedFlow<ConfigUpdate>(
replay = 5, // 缓存最近5次配置变更
extraBufferCapacity = 20, // 支持大量订阅者
onBufferOverflow = BufferOverflow.DROP_OLDEST // 配置更新以最新为准
)

val configFlow: SharedFlow<ConfigUpdate> = _configFlow.asSharedFlow()

// 批量更新配置
suspend fun updateMultipleConfigs(configs: List<ConfigUpdate>) {
configs.forEach { config ->
_configFlow.emit(config)
}
}
}

// 配置更新事件
data class ConfigUpdate(
val key: String,
val value: Any,
val version: Int
)

这种配置特别适合需要追踪配置变更历史的场景，新订阅者可获取最近的配置变更记录。