PublishSubject 与 Kotlin 协程(Flow)

发布于05月13日

我使用了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 */ }
}

Since I need the debounce operator I really wanted to do the same thing with flows so I created a channel and then I tried to create a flow from that channel and listen to changes, but I'm not getting any results.

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
    }
}

怎么了？

推荐答案

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
    }
}

_{Update:问题提出时，有两个参数(如问题中所示)的过载为debounce.再也没有了.但是现在有一个以毫秒(长)为单位的参数}