PublishSubject 与 Kotlin 协程（流程）

Question

我使用了一个 PublishSubject，我正在向它发送消息，同时我也在监听结果。它完美无缺，但现在我不确定如何使用 Kotlin 的协程（流或通道）做同样的事情。

private val subject = PublishProcessor.create<Boolean>>()

...

fun someMethod(b: Boolean) {
    subject.onNext(b)
}

fun observe() {
    subject.debounce(500, TimeUnit.MILLISECONDS)
           .subscribe { /* value received */ }
}

由于我需要 debounce 运算符，我真的很想对流做同样的事情，所以我创建了一个通道，然后我尝试从该通道创建一个流并监听变化，但我没有得到任何结果。

private val channel = Channel<Boolean>()

...

fun someMethod(b: Boolean) {
    channel.send(b)
}

fun observe() {
    flow {
         channel.consumeEach { value ->
            emit(value)
         }
    }.debounce(500, TimeUnit.MILLISECONDS)
    .onEach {
        // value received
    }
}

怎么了？

Answer 1

Flow 是冷异步流，就像Observable。

流上的所有转换，例如map 和filter 都不会触发流收集或执行，只有终端操作符（例如single）会触发它。

onEach 方法只是一种转换。因此，您应该将其替换为终端流运算符collect。您也可以使用BroadcastChannel 来获得更简洁的代码：

private val channel = BroadcastChannel<Boolean>(1)

suspend fun someMethod(b: Boolean) {
    channel.send(b)
}

suspend fun observe() {
  channel
    .asFlow()
    .debounce(500)
    .collect {
        // value received
    }
}

_{更新：在提出问题时，debounce 有两个参数（如问题中所示）的过载。没有了。但是现在有一个以毫秒为单位的参数（长）。}

Answer 2

对于PublishProcessor/PublishRelay，应该是SharedFlow/MutableSharedFlow

private val _myFlow = MutableSharedFlow<Boolean>(
                          replay = 0,
                          extraBufferCapacity = 1, // you can increase      
                          BufferOverflow.DROP_OLDEST
)
val myFlow = _myFlow.asSharedFlow()


// ...
fun someMethod(b: Boolean) {
    _myFlow.tryEmit(b)
}

fun observe() {
    myFlow.debounce(500)
          .onEach {  }
          // flowOn(), catch{}
          .launchIn(coroutineScope)

}

还有StateFlow/MutableStateFlow 代表BehaviorProcessor/BehaviorRelay。

private val _myFlow = MutableStateFlow<Boolean>(false)
val myFlow = _myFlow.asStateFlow()

// ...
fun someMethod(b: Boolean) {
    _myFlow.value = b // same as _myFlow.emit(v), myFlow.tryEmit(b)
}

fun observe() {
    myFlow.debounce(500)
          .onEach {  }
          // flowOn(), catch{}
          .launchIn(coroutineScope)

}

StateFlow 必须有初始值，如果你不想这样，这是解决方法：

private val _myFlow = MutableStateFlow<Boolean?>(null)
val myFlow = _myFlow.asStateFlow()
                    .filterNotNull()

MutableStateFlow 在设置新值时使用.equals 比较，因此它不会一次又一次地发出相同的值（相对于使用引用比较的distinctUntilChanged）。

所以MutableStateFlow ≈ BehaviorProcessor.distinctUntilChanged()。如果你想要精确的BehaviorProcessor 行为，那么你可以使用这个：

private val _myFlow = MutableSharedFlow<Boolean>(
                              replay = 1, 
                              extraBufferCapacity = 0,
                              BufferOverflow.DROP_OLDEST
)

Answer 3

Kotlin 协程中的 ArrayBroadcastChannel 与 PublishSubject 最相似。

1. 和 PublishSubject 一样，一个 ArrayBroadcastChannel 可以有多个 订阅者和所有活动订阅者都会立即收到通知。
2. 与 PublishSubject 一样，如果目前没有活动订阅者，推送到此频道的事件会丢失。

与 PublishSubject 不同，背压内置于协程通道中，这就是缓冲容量的来源。这个数字实际上取决于通道用于哪个用例。对于大多数正常用例，我只选择 10 个，这应该绰绰有余。如果您将事件推送到此通道的速度比接收它的接收器快，您可以提供更多容量。

Answer 4

实际上 BroadcastChannel 已经过时了，Jetbrains 改变了他们的方法，改为使用 SharedFlows。这更清洁，更易于实施并解决了很多痛点。

基本上，你可以实现同样的目标。

class BroadcastEventBus {
    private val _events = MutableSharedFlow<Event>()
    val events = _events.asSharedFlow() // read-only public view

    suspend fun postEvent(event: Event) {
        _events.emit(event) // suspends until subscribers receive it
    }
}

要了解更多信息，请查看 Roman 的 Medium 文章。

"Shared flows, broadcast channels" by Roman Elizarov