RXJAVA2如果元素当前存在于流中，则忽略它

Question

我有一个情况，如果我需要删除/忽略元素，如果它是当前在反应流中呈现的。例如：

fun ignoreDuplicatesExample() {
    val publishSubject: PublishSubject<Long> = PublishSubject.create()
    publishSubject.observeOn(Schedulers.single()).distinct().subscribe({
        Thread.sleep(1000)
        println("onNext: $it")
    }, {
        error("$it")
    })
    publishSubject.onNext(1)
    publishSubject.onNext(2)
    publishSubject.onNext(3)
    publishSubject.onNext(1) // should be ignored
    publishSubject.onNext(2) // should be ignored
    publishSubject.onNext(3) // should be ignored
    Thread.sleep(10_000)
    publishSubject.onNext(1) // by this time it should be already consumed, so it need to be allowed to emit it again
    publishSubject.onNext(4)
    Thread.sleep(10_000)
    println("exit")
}

输出：

onNext: 1
onNext: 2
onNext: 3
onNext: 4
exit

，但我希望看到：

onNext: 1
onNext: 2
onNext: 3
onNext: 1
onNext: 4
exit

所以，有人知道如何使用rxjava2？

Answer 1

您不能直接地做到这一点，因为上流链不应该在流中不知道，如果它是"消耗" d。(想象一下是否有几个订户。(如果您想这样做，则需要与流中外部的变量进行交互。