企业官网建设流程全解析

一、概念

1.1 通信模式

根据接收端节点数量以及接收方式，可分为三类。

1.2 扇入扇出

二、协程间通信 Channel

出现在Flow之前，现在退居幕后职责单一，仅作为协程间通信的并发安全的缓冲队列而存在。两个消费者从同一个队列里取数据，一个取走一条，另一个再取下一条，它们是在分摊消费，不是在共享订阅。

SendChannel在创建时定义了消费方式外界只能往里发送值，ReceiveChannel在创建时定义了生产方式外界只能从中获取值，Channel继承了它俩既能发送也能接收，根据实际需求暴露不同类型收窄功能。

• 非阻塞：类似于 Java 中的 BlockingQueue 队列，不同的是 put() 和 take() 读取写入数据是阻塞的，而 Channel 中的 send() 和 receive() 是挂起的。
• 同步性：每个值只能被众多订阅者中的一个消费。
• 并发安全：没有检测到 receive() 的话 send() 就会挂起不会发送值（默认模式 RENDEZVOUS，没有缓冲区的Channel是同步的）。
• 公平性：在多个协程中发送或接收（多线程竞争）遵循先进先出（FIFO即队列）。

• send() 和 receive()：都是挂起函数。一个协程往队列里放数据，另一个协程从队列里取数据。队列为空时接收方挂起，队列满时发送方挂起。
• trySend() 和 tryReceive()：普通函数。从非挂起函数中发送或接收元素，操作是即时的并返回 ChannelResult 对象，包含了有关操作成功或失败的信息。能发就发，发不了就立刻返回，能收就收，收不到也立刻返回。绝不等待，不像挂起版会挂起等待来确保完成操作。
• close() 和 cancel()：当创建的是 Channel 类型的时候，两个都可以用来关闭。close() 是从生产端关闭，表示不再需要新数据，会消费完缓冲区中已有的元素。cancel() 是从消费端取消，表示不再需要数据，会直接关闭并丢未消费的元素（若元素持有资源，考虑通过 onUndeliveredElement() 处理这些被丢弃的元素进行收尾）。通道使用完一定要关闭，否则代码块不会停止，造成协程阻塞和内存泄漏。
• send() 和 trySend()：由于 emit() 会等待收集器处理完数据，在 collect 处理数据的这 5 秒内，emit() 所在的协程会被阻塞，无法继续执行后续的 emit() 操作，这可能会影响整个程序的性能和响应速度 。因此，在使用 emit() 时，需要充分考虑收集器的性能，确保不会因为收集器的问题导致 emit() 长时间阻塞，进而影响其他操作的执行 。tryEmit() 会尝试立即发送数据，如果发送失败（例如缓冲区已满），不会进行重试或等待，直接返回 false。这就像在网络信号不稳定时发送消息，发送者尝试发送后，不管消息是否真正送达，就继续做其他事情 。在一些对数据实时性要求不高，允许部分数据丢失的场景，如日志记录、实时监控数据的快速上报等，tryEmit 的这种策略能提高系统的性能和响应速度 。

fun main() = runBlocking { private val _channel = Channel<String>() val receiveChannel: ReceiveChannel<String> = _channel val sendChannel: SendChannel<String> = _channel _channel.send("Channel发送") sendChannel.send("SendChannel发送") delay(1000) _channel.consumeEach { println("Channel接收：$it") } receiveChannel.consumeEach { println("ReceiveChannel接收：$it") } }

2.1 创建

produce() 和 actor() 被定义成协程构建器（因此只能在协程环境中调用，会在异常、完成、取消时自动关闭），同 launch、async 一样作为 CoroutineScope 的扩展函数。

• actor() 在 kotlinx.coroutines 1.0.0（2018年10月）被标记为 ObsoleteCoroutinesApi，处于不推荐使用也暂时不会消失的状态。
• produce() 依然可用，但 Flow 通常是更现代、更主流的选择。例如 ChannelFlow 或 CallbackFlow 来构建异步数据流。

2.1.1 actor() 创建 SendChannel

创建时在协程构建器 actor() 的 Lambda 中定义了数据的消费方式，返回一个生产者通道 SendChannel，其它协程通过该对象往里发送数据。把共享状态藏进 actor() 里，外部协程只能往里发消息，actor() 内部串行处理，天然没有竞态。

fun main() = runBlocking { //创建生产者通道 val send: SendChannel<Int> = actor { for (i in channel) println(i) //定义了消费方式 } //其它协程拿来发送数据 launch { (1..3).forEach { sendChannel.send(it) } } }

2.1.2 produce() 创建 ReceiveChannel

创建时在协程构建器 produce() 的 Lambda 中定义了数据的生产方式，返回一个消费者通道 ReceiveChannel，其它协程通过该对象从中取出数据。

fun main() = runBlocking { //创建消费者通道 val receiveChannel: ReceiveChannel<Int> = produce { (1..3).forEach { send(it) } //定义了生产方式 } //其它协程拿来接收数据 launch { for (i in receiveChannel) println(i) } }

2.1.3 构造创建 Channel

val rendezvousChannel = Channel<String>() //约会类型 val bufferedChannel = Channel<String>(10) //指定缓存大小类型 val conflatedChannel = Channel<String>(Channel.CONFLATED) //混合类型 val unlimitedChannel = Channel<String>(Channel.UNLIMITED) //无限缓存大小类型

fun main() = runBlocking { val channel = Channel<Int>(capacity = Channel.CONFLATED) { println("onUndeliveredElement: $it") } launch { (1..3).forEach { println("send: $it") channel.send(it) } channel.close() //不要忘记关闭 } launch { for (i in channel) { println("receive: $i") } } println("结束！") } 打印： 结束！ send: 1 send: 2 onUndeliveredElement: 1 send: 3 onUndeliveredElement: 2 receive: 3

2.2 遍历元素

接收端生命周期结束时，考虑是否应该调用 cancel()。如果有未交付元素持有资源，再考虑是否需要 onUndeliveredElement 做收尾。

2.2.1 Iterate 迭代器

可以理解为不断调用 receive()，没有元素时会挂起等待，直到有新元素，或者通道关闭。

fun main(): Unit = runBlocking { val channel = Channel<String>(Channel.UNLIMITED) repeat(5) { channel.send("$it") } val iterator = channel.iterator() while (iterator.hasNext()) { println("【iterator】${iterator.next()}") delay(1000) } //5个元素打印完后程序没结束，Channel不会关闭，后面代码执行不到 println("这行执行不到") }

2.2.2 for 循环

可以理解为不断调用 receive()，没有元素时会挂起等待，直到有新元素，或者通道关闭。

fun main():Unit = runBlocking { val channel = Channel<String>(Channel.UNLIMITED) repeat(5) { channel.send("$it") } for (i in channel) { println("【for】$i,") delay(1000) } //5个元素打印完后程序没结束，Channel不会关闭，后面代码执行不到 println("这行执行不到") }

2.2.3 扩展函数

会在代码块执行完后关闭通道，但无法保证在代码块执行完后不会有新元素进入，新元素会被丢弃，可在通过构造方式创建通道时指定 onUndeliveredElement() 进行处理。

//只消费第一个元素并关闭通道 suspend fun <E> ReceiveChannel<E>.consumeFirst(): E = consume { return receive() }

2.2.4 转换成 Flow

底层通过 ChannelFlow 实现，Channel 转成 Flow 是为了使用那些方便的操作符，同时 Flow 的很多功能扩展底层由 Channel 实现（如flowOn、buffer）。

三、事件流SharedFlow

BroadcastChannel 在协程1.4版本已被废弃，取而代之的是 SharedFlow（StateFlow是它的特定配置版本）。

SharedFlow只能订阅（消费），FlowCollector只能发送（生产），MutableSharedFlow继承了它俩既能发送也能订阅，根据实际需求暴露不同类型收窄功能。接收会一直监听，通过取消协程来关闭它的收集。

fun main() = runBlocking { //利用多态暴露不同父类限制功能给外部使用 private val _mutableSharedFlow = MutableSharedFlow<String>() val sharedFlow: SharedFlow<String> = _mutableSharedFlow //或调用 asSharedFlow() val flowCollector: FlowCollector<String> = _mutableSharedFlow launch { _mutableSharedFlow.collect { println("mutableSharedFlow接收：$it") } } launch { sharedFlow.collect { println("mutableSharedFlow接收：$it") } } delay(1000) mutableSharedFlow.emit("mutableSharedFlow发送") flowCollector.emit("flowCollector发送") }

3.1 通过构造创建

3.1.1 回放

新订阅者能不能拿到订阅前已经播过的旧值，能拿几个。

3.1.2 额外缓存容量

在回放之外额外增加缓存量，所以 replay + extraBufferCapacity = 才是总共的缓存数量。生产者最多可以额外缓冲几个值，而不必马上挂起（如果缓存溢出策略设置的是挂起的话）。

3.1.3 缓存溢出策略

缓冲区满了以后（事件太多出现背压），是挂起、丢旧的、还是丢新的。只有存在消费者时才会触发缓存溢出策略，否则直接丢弃。只有在回放或缓存容量>0时，才支持两种丢弃模式。

例如两个launch虽然会并行执行，若下方的消费launch在上方的生产launch后执行，先发送的值都会被丢弃接收不到，直到消费launch执行了才开始接收后面的值。这是和 Channel 最大的区别，没有消费者 Channel 不会丢。

suspend fun method(): Unit = coroutineScope { //让3个launch同时进行 val shared = MutableSharedFlow<Int>(3) //回放3个数据 launch { for(i in 1..5){ shared.emit(i) println("emmit：$i") //如果这里不延迟，虽然launch是并行执行，发送是很快的这里才5个值 //相当于是没有订阅者的，5个值都会被丢弃 //除非把下面的订阅launch写在这个launch上面 delay(1000) //1秒更新一个值 } } //订阅者甲 launch { shared.collect{ println("甲：$it") } } //订阅者乙5秒后再订阅，也就是值全都更新完了再订阅 delay(5000) launch { shared.collect{ println("乙：$it") } } } 打印： emmit：1 甲：1 emmit：2 甲：2 emmit：3 甲：3 emmit：4 甲：4 emmit：5 甲：5 乙：3 //回放了3个已更新过的数据 乙：4 乙：5

3.2 Flow 转 SharedFlow

四、状态流 StateFlow

StateFlow继承自SharedFlow是一种特殊配置，相当于MutableSharedFlow(1,0, BufferOverflow.DROP_OLDEST)，可以使用value属性来访问值，可以当作是用来取代LiveData。

• 必须传入默认值：Null安全。
• 回放个数1+额外缓冲区大小0：只持有1个值。
• 缓存策略DROP_OLDEST：只持有最新值（新值会替换旧值）。
• 数据防抖：即仅在新值内容发生变化才会消费。

4.1 通过构造创建

val state = MutableStateFlow(0) //默认值会被覆盖 launch { for(i in 1..5){ state.emit(i) println("emit：$i") delay(1000) //1秒更新一个值 } } launch { delay(2000) //2秒后开始订阅 state.collect{ println("collect：$it") } } 打印： emit：1 emit：2 collect：2 //2秒后开始订阅，不会收到之前已更新的数据 emit：3 collect：3 emit：4 collect：4 emit：5 collect：5

4.2 Flow 转 StateFlow

val uiState: StateFlow<UiState> = repository.getDataFlow() .onStart { emit(UiState.Loading) } .stateIn( scope = viewModelScope, started = SharingStarted.WhileSubscribed(5_000), initialValue = UiState.Loading )

四、区别及选择

4.1 通信(Channel) or 广播(SharedFlow)

4.2 事件(Event) or 状态(State)

事件按顺序都要执行到，状态只关心最新值。