我有一个类(使用线程池)不断生成和存储新的数据值。我想为客户端代码("订阅者")提供一种方法，以利用(连接到)新数据值序列。但是，如果我的类没有客户端，或者所有客户端都已完成对序列的读取，我希望它继续生成和存储新值，而不停止。当客户端连接到该序列时，它会接收新生成的值，但不会接收过去生成的值。哪个项目反应堆类别适合这样做？
我想我需要使用Flux来表示新值的序列，但要使用哪个Flux类(或工厂方法)？
使用DirectProcessor
据我所知，所需要的是订阅上游的能力，无论是否有其订户。
这可以在DirectProcessor的支持下实现。由于Processor是Publisher和Subscriber的组合，因此它可以向上游"运行"并连续监听传入信号。同时，DirectProcessor启用消息解复用，或者简单地向所有可用的下游用户广播消息(如果他们正在侦听)。
例如，让我们考虑以下代码示例：
Flux<Long> intervalFlux = Flux.interval(Duration.ofMillis(500)).log("upstream");
DirectProcessor processor = DirectProcessor.create();
intervalFlux.subscribe(processor);
Thread.sleep(2000);
Disposable downstream1 = processor.log("downstream1")
.subscribe();
Thread.sleep(1000);
downstream1.dispose();
Thread.sleep(1000);
Disposable downstream2 = processor.log("downstream2")
.subscribe();
Thread.sleep(2000);
正如我们所看到的，我们使用处理器订阅了上游，因此间隔Flux开始生成数据。然后，我们订阅了processor并等待1秒，因此下行流1和6应该观察到两个事件，log("upstream")操作员通常会记录这两个事件。之后，我们取消了订阅，因此downstream1订阅方应该停止观察任何事件，但log("upstream")应该仍然观察间隔。然后，在又一次暂停之后，我们与另一个downstrea2订阅者订阅了流，该订阅者应该观察另外四个事件。
上述代码的一般输出如下：
2019-01-23 15:09:04,246 INFO upstream [main] onSubscribe(FluxInterval.IntervalRunnable)
2019-01-23 15:09:04,249 INFO upstream [main] request(unbounded)
2019-01-23 15:09:04,757 INFO upstream [parallel-1] onNext(0)
2019-01-23 15:09:05,252 INFO upstream [parallel-1] onNext(1)
2019-01-23 15:09:05,751 INFO upstream [parallel-1] onNext(2)
2019-01-23 15:09:06,252 INFO upstream [parallel-1] onNext(3)
2019-01-23 15:09:06,258 INFO downstream1 [main] onSubscribe(DirectProcessor.DirectInner)
2019-01-23 15:09:06,258 INFO downstream1 [main] request(unbounded)
2019-01-23 15:09:06,754 INFO upstream [parallel-1] onNext(4)
2019-01-23 15:09:06,755 INFO downstream1 [parallel-1] onNext(4)
2019-01-23 15:09:07,254 INFO upstream [parallel-1] onNext(5)
2019-01-23 15:09:07,254 INFO downstream1 [parallel-1] onNext(5)
2019-01-23 15:09:07,263 INFO downstream1 [main] cancel()
2019-01-23 15:09:07,755 INFO upstream [parallel-1] onNext(6)
2019-01-23 15:09:08,255 INFO upstream [parallel-1] onNext(7)
2019-01-23 15:09:08,265 INFO downstream2 [main] onSubscribe(DirectProcessor.DirectInner)
2019-01-23 15:09:08,265 INFO downstream2 [main] request(unbounded)
2019-01-23 15:09:08,755 INFO upstream [parallel-1] onNext(8)
2019-01-23 15:09:08,756 INFO downstream2 [parallel-1] onNext(8)
2019-01-23 15:09:09,255 INFO upstream [parallel-1] onNext(9)
2019-01-23 15:09:09,256 INFO downstream2 [parallel-1] onNext(9)
2019-01-23 15:09:09,751 INFO upstream [parallel-1] onNext(10)
2019-01-23 15:09:09,751 INFO downstream2 [parallel-1] onNext(10)
2019-01-23 15:09:10,255 INFO upstream [parallel-1] onNext(11)
2019-01-23 15:09:10,255 INFO downstream2 [parallel-1] onNext(11)
正如我们所看到的，DirectProcessor启用了所需的行为，因此它可能很适合那里。
备注
DirectProcessor不支持背压，因此在背压很重要的情况下，可以在那里使用limitRate运算符。
另请参阅
https://projectreactor.io/docs/core/release/reference/#_direct_processor