如何在不关闭流的情况下从流量中收集

Question

我的做法是创建一个反应endpoint，如下所示：

public Flux<ServerEvent> getEventFlux(Long forId){
    ServicePoller poller = new ServicePollerImpl();
    Map<String,Object> params =  new HashMap<>();
    params.put("id", forId);

    Flux<Long> interval = Flux.interval(Duration.ofMillis(pollDuration));
    Flux<ServerEvent> serverEventFlux =
            Flux.fromStream(
                    poller.getEventStream(url, params) //poll a given endpoint after a fixed duration.
            );
    Flux<ServerEvent> sourceFlux= Flux.zip(interval, serverEventFlux)
             .map(Tuple2::getT2); // Zip the two streams. 
/* Here I want to store data from sourceFlux into a collection whenever some data arrives without disturbing the downstream processing in Spring. So that I can access collection later on without polling again */

这会在数据可用时立即将其发回前端，然而，我的第二个用例是在数据到达时将其汇集到一个单独的集合中，这样，如果以后有类似的请求到达，我就可以从池中卸载整个数据，而不必再次访问服务。

在不关闭流量流的情况下，我有什么选择来访问流量并在它们到达时将值存储到集合中？

遇到异常：

java.lang.IllegalStateException:stream已经在java.util.stream.abstractPipeline.spliterator（abstractPipeline.java:343)~[NA:1.8.0_171]在java.util.stream.referencePipeline.iterator（referencePipeline.java:139)~[NA:1.8.0_171]在reactor.core.publisher.fluxstream.subscribe（jar:3.1.7.发布]在reactor.core.publisher.fluxzip$zipcoordator.subscribe（fluxzip.java:573)~[reactor-core-3.1.7.release.jar:3.1.7.发布]在reactor.core.publisher.fluxzip.handleboth(fluxzip.java:326)~[reactor-core-3.1.7.release.jar:3.1.7.发布]在reactor.core.publisher.fluxzip.jar:3.1.7.发布]

Answer 1

poller.getEventStream返回一个只能使用一次的Java8流。您可以先将流转换为集合，或者通过使用供应商推迟poller.getEventStream的执行：

Flux.fromStream(
  () -> poller.getEventStream(url, params)
);