为什么 Sinks.many().multicast().onBackPressureBuffer() 在订阅者之一取消订阅后完成以及如何避免它

我在使用时遇到了我不明白的行为Sinks.Many<String>向多个订阅者通知某些事件：

fun main() {

    val sink : Sinks.Many<String>  = Sinks.many().multicast().onBackpressureBuffer()
    val flux = sink.asFlux().log()

    val d = flux.subscribe {
        println("--> $it")
    }

    sink.emitNext("1", Sinks.EmitFailureHandler.FAIL_FAST)

    val d2 = flux.subscribe {
        println("--2> $it")
    }

    sink.emitNext("2", Sinks.EmitFailureHandler.FAIL_FAST)
}

此代码显示第一个订阅者获取值 1 和 2，第二个订阅者获取值 2。到目前为止一切顺利：

11:49:06.936 [main] INFO reactor.Flux.EmitterProcessor.1 - onSubscribe(EmitterProcessor.EmitterInner)
11:49:06.938 [main] INFO reactor.Flux.EmitterProcessor.1 - request(unbounded)
11:49:06.942 [main] INFO reactor.Flux.EmitterProcessor.1 - onNext(1)
--> 1
11:49:06.942 [main] INFO reactor.Flux.EmitterProcessor.1 - onSubscribe(EmitterProcessor.EmitterInner)
11:49:06.942 [main] INFO reactor.Flux.EmitterProcessor.1 - request(unbounded)
11:49:06.943 [main] INFO reactor.Flux.EmitterProcessor.1 - onNext(2)
--> 2
11:49:06.943 [main] INFO reactor.Flux.EmitterProcessor.1 - onNext(2)
--2> 2

现在，假设第一个订阅者在第一次发射后处置（取消）其订阅，我期望第一个订阅者获得 1，第二个订阅者获得 2：

    val sink : Sinks.Many<String>  = Sinks.many().multicast().onBackpressureBuffer()
    val flux = sink.asFlux().log()

    val d = flux.subscribe {
        println("--> $it")
    }

    sink.emitNext("1", Sinks.EmitFailureHandler.FAIL_FAST)

    d.dispose()

    val d2 = flux.subscribe {
        println("--2> $it")
    }

    sink.emitNext("2", Sinks.EmitFailureHandler.FAIL_FAST)

}

11:51:48.684 [main] INFO reactor.Flux.EmitterProcessor.1 - onSubscribe(EmitterProcessor.EmitterInner)
11:51:48.685 [main] INFO reactor.Flux.EmitterProcessor.1 - request(unbounded)
11:51:48.689 [main] INFO reactor.Flux.EmitterProcessor.1 - onNext(1)
--> 1
11:51:48.689 [main] INFO reactor.Flux.EmitterProcessor.1 - cancel()
11:51:48.689 [main] INFO reactor.Flux.EmitterProcessor.1 - onSubscribe(EmitterProcessor.EmitterInner)
11:51:48.689 [main] INFO reactor.Flux.EmitterProcessor.1 - request(unbounded)
11:51:48.690 [main] INFO reactor.Flux.EmitterProcessor.1 - onComplete()

然而，当第二个订阅者尝试订阅时，通量被视为已完成。为什么会发生这种情况？我需要 Sinks.Many 可以随时订阅和取消订阅，而无需取消。

我刚刚遇到了同样的问题。

这是由 autoCancel 默认为 true 引起的。不幸的是onBackPressureBuffer javadoc https://projectreactor.io/docs/core/3.4.0-SNAPSHOT/api/reactor/core/publisher/Sinks.MulticastSpec.html#onBackpressureBuffer--没有提及它。

这种行为继承自EmitterProcessor.create https://projectreactor.io/docs/core/3.4.0-SNAPSHOT/api/reactor/core/publisher/EmitterProcessor.html#create--它被记录在哪里。

要将 autoCancel 标志设置为 false，需要使用替代方法背压缓冲区 https://projectreactor.io/docs/core/3.4.0-SNAPSHOT/api/reactor/core/publisher/Sinks.MulticastSpec.html#onBackpressureBuffer-int-boolean-

Many<String> sink = Sinks.many().multicast().onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, false);