val coldFlow: Flow<Int> = flow {
    // 生产者代码
    while(true) {
        // 执行计算
        emit(result)
        delay(100)
    }
    // 生产者代码结束，流将被关闭
}.collect{ data ->  
}
复制代码

public abstract class AbstractFlow<T> : Flow<T> {
    @InternalCoroutinesApi
    public final override suspend fun collect(collector: FlowCollector<T>) {
        // 1. 对 flow{} 的包装
        val safeCollector = SafeCollector(collector, coroutineContext)
        try {
            // 2. 执行 flow{} 代码块
            collectSafely(safeCollector)
        } finally {
            // 3. 释放协程相关的参数
            safeCollector.releaseIntercepted()
        }
    }
    public abstract suspend fun collectSafely(collector: FlowCollector<T>)
}
private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
    override suspend fun collectSafely(collector: FlowCollector<T>) {
        collector.block()
    }
}
复制代码

public interface SharedFlow<out T> : Flow<T> {
    // 缓存的重放数据的快照
    public val replayCache: List<T>
}
public interface MutableSharedFlow<T> : SharedFlow<T>, FlowCollector<T> {
    // 发射数据（注意这是个挂起函数）
    override suspend fun emit(value: T)
    // 尝试发射数据（如果缓存溢出策略是 SUSPEND，则溢出时不会挂起而是返回 false）
    public fun tryEmit(value: T): Boolean
    // 活跃订阅者数量
    public val subscriptionCount: StateFlow<Int>
    // 重置重放缓存，新订阅者只会收到注册后新发射的数据
    public fun resetReplayCache()
}
复制代码

public fun <T> MutableSharedFlow(
    // 重放数据个数
    replay: Int = 0,
    // 额外缓存容量
    extraBufferCapacity: Int = 0,
    // 缓存溢出策略
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
): MutableSharedFlow<T> {
    val bufferCapacity0 = replay + extraBufferCapacity
    val bufferCapacity = if (bufferCapacity0 < 0) Int.MAX_VALUE else bufferCapacity0 // coerce to MAX_VALUE on overflow
    return SharedFlowImpl(replay, bufferCapacity, onBufferOverflow)
}
public enum class BufferOverflow {
    // 挂起
    SUSPEND,
    // 丢弃最早的一个
    DROP_OLDEST,
    // 丢弃最近的一个
    DROP_LATEST
}
复制代码

public fun <T> Flow<T>.shareIn(
    // 协程作用域范围
    scope: CoroutineScope,
    // 启动策略
    started: SharingStarted,
    // 控制数据重放的个数
    replay: Int = 0
): SharedFlow<T> {
  val config = configureSharing(replay)
  val shared = MutableSharedFlow<T>(
      replay = replay,
      extraBufferCapacity = config.extraBufferCapacity,
      onBufferOverflow = config.onBufferOverflow
  )
  @Suppress("UNCHECKED_CAST")
  scope.launchSharing(config.context, config.upstream, shared, started, NO_VALUE as T)
  return shared.asSharedFlow()
}
public companion object {
    // 热启动式：立即开始，并在 scope 指定的作用域结束时终止
    public val Eagerly: SharingStarted = StartedEagerly()
    // 懒启动式：在注册首个订阅者时开始，并在 scope 指定的作用域结束时终止
    public val Lazily: SharingStarted = StartedLazily()
    public fun WhileSubscribed(
        stopTimeoutMillis: Long = 0,
        replayExpirationMillis: Long = Long.MAX_VALUE
    ): SharingStarted =
        StartedWhileSubscribed(stopTimeoutMillis, replayExpirationMillis)
}
复制代码

public interface StateFlow<out T> : SharedFlow<T> {
    // 当前值
    public val value: T
}
public interface MutableStateFlow<T> : StateFlow<T>, MutableSharedFlow<T> {
    // 当前值
    public override var value: T
    // 比较并设置（通过 equals 对比，如果值发生真实变化返回 true）
    public fun compareAndSet(expect: T, update: T): Boolean
}
复制代码

public fun <T> MutableStateFlow(value: T): MutableStateFlow<T> = StateFlowImpl(value ?: NULL)
复制代码

val shared = MutableSharedFlow(
    replay = 1,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)
shared.tryEmit(initialValue) // emit the initial value
val state = shared.distinctUntilChanged() // get StateFlow-like behavior
复制代码

public override var value: T
    get() = NULL.unbox(_state.value)
    set(value) { updateState(null, value ?: NULL) }
override fun compareAndSet(expect: T, update: T): Boolean =
    updateState(expect ?: NULL, update ?: NULL)
private fun updateState(expectedState: Any?, newState: Any): Boolean {
    var curSequence = 0
    var curSlots: Array<StateFlowSlot?>? = this.slots // benign race, we will not use it
    synchronized(this) {
        val oldState = _state.value
        if (expectedState != null && oldState != expectedState) return false // CAS support
        if (oldState == newState) return true // 如果新值 equals 旧值则拦截, 但 CAS 返回 true
        _state.value = newState
        ...
        return true
    }
}
复制代码

public fun <T> Flow<T>.stateIn(
    // 共享开始时所在的协程作用域范围
    scope: CoroutineScope,
    // 共享开始策略
    started: SharingStarted,
    // 初始值
    initialValue: T
): StateFlow<T> {
    val config = configureSharing(1)
    val state = MutableStateFlow(initialValue)
    scope.launchSharing(config.context, config.upstream, state, started, initialValue)
    return state.asStateFlow()
}
复制代码

// 从 2.4.0 开始支持 Lifecycle#repeatOnLifecycle API
implementation "androidx.lifecycle:lifecycle-runtime-ktx:2.4.1"
复制代码

class LocationActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        lifecycleOwner.addRepeatingJob(Lifecycle.State.STARTED) {
            locationProvider.locationFlow().collect {
                // update UI
            }
        }
    }
}
class LocationActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        // repeatOnLifecycle 是 suspends 函数，所以需要在协程中执行
        // 当 lifecycleScope 的生命周期高于 STARTED 状态时，启动一个新的协程并执行代码块
        // 当 lifecycleScope 的生命周期低于 STARTED 状态时，取消该协程
        lifecycleScope.launch {
            repeatOnLifecycle(Lifecycle.State.STARTED) {
                // 当前生命周期一定高于 STARTED 状态，可以安全地从数据流中取数据，并更新 View
                locationProvider.locationFlow().collect {
                    // update UI
                }
            }
        // 结构化并发：生命周期处于 DESTROYED 状态时，切换回调用 repeatOnLifecycle 的协程继续执行
        }
    }
}
class LocationActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        locationProvider.locationFlow()
            .flowWithLifecycle(this, Lifecycle.State.STARTED)
            .onEach {
                // update UI
            }
            .launchIn(lifecycleScope) 
    }
}
复制代码

class LocationActivity : AppCompatActivity() {
    // 协程控制器
    private var locationUpdatesJob: Job? = null
    override fun onStart() {
        super.onStart()
        locationUpdatesJob = lifecycleScope.launch {
            locationProvider.locationFlow().collect {
                // update UI
            } 
        }
    }
    override fun onStop() {
       // 在视图进入后台时取消协程
        locationUpdatesJob?.cancel()
        super.onStop()
    }
}
复制代码

suspendCancellableCoroutine<Unit> { cont ->
    // Lifecycle observers that executes `block` when the lifecycle reaches certain state, and
    // cancels when it falls below that state.
    val startWorkEvent = Lifecycle.Event.upTo(state)
    val cancelWorkEvent = Lifecycle.Event.downFrom(state)
    val mutex = Mutex()
    observer = LifecycleEventObserver { _, event ->
        if (event == startWorkEvent) {
            // Launch the repeating work preserving the calling context
            launchedJob = this@coroutineScope.launch {
                // Mutex makes invocations run serially,
                // coroutineScope ensures all child coroutines finish
                mutex.withLock {
                    coroutineScope {
                        block()
                    }
                }
            }
            return@LifecycleEventObserver
        }
        if (event == cancelWorkEvent) {
            launchedJob?.cancel()
            launchedJob = null
        }
        if (event == Lifecycle.Event.ON_DESTROY) {
            cont.resume(Unit)
        }
    }
    this@repeatOnLifecycle.addObserver(observer as LifecycleEventObserver)
}
复制代码

public fun <E> Channel(
    // 缓冲区容量，当超出容量时会触发 onBufferOverflow 拒绝策略
    capacity: Int = RENDEZVOUS,  
    // 缓冲区溢出策略，默认为挂起，还有 DROP_OLDEST 和 DROP_LATEST
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    // 处理元素未能成功送达处理的情况，如订阅者被取消或者抛异常
    onUndeliveredElement: ((E) -> Unit)? = null
): Channel<E>
复制代码

interface UiState
interface UiEvent
interface UiEffect
abstract class BaseViewModel<State : UiState, Event : UiEvent, Effect : UiEffect> : ViewModel() {
    // 初始状态
    private val initialState: State by lazy { createInitialState() }
    // 页面需要的状态，对应于 MVI 模式的 ViewState
    private val _uiState = MutableStateFlow<State>(initialState)
    // 对外接口使用不可变版本
    val uiState = _uiState.asStateFlow()
    // 页面状态变更的 “副作用”，类似一次性事件，不需要重放的状态变更（例如 Toast）
    private val _effect = MutableSharedFlow<Effect>()
    // 对外接口使用不可变版本
    val effect = _effect.asSharedFlow()
    // 页面的事件操作，对应于 MVI 模式的 Intent 
    private val _event = MutableSharedFlow<Event>()
    init {
        viewModelScope.launch {
            _event.collect {
                handleEvent(it)
            }
        }
    }
    // 初始状态
    protected abstract fun createInitialState(): State
    // 事件处理
    protected abstract fun handleEvent(event: Event)
    /**
     * 事件入口
     */
    fun sendEvent(event: Event) {
        viewModelScope.launch {
            _event.emit(event)
        }
    }
    /**
     * 状态变更
     */
    protected fun setState(newState: State) {
        _uiState.value = newState
    }
    /**
     * 副作用
     */
    protected fun setEffect(effect: Effect) {
        _effect.send(effect)
    }
}