Spring WebFlux（Flux）：如何动态发布

Question

我是反应编程和Spring WebFlux的新手。我想让我的应用程序1发布服务器发送的事件通过Flux和我的应用程序2不断地监听它。

我希望Flux按需发布（例如，当某些事情发生时）。我发现的所有示例都是使用Flux.Interval周期性地发布event，而且似乎没有办法在创建完Flux之后追加/修改其中的内容。

我怎样才能达到我的目标？或者我在概念上完全错了。

Answer 1

手动向Flux提供数据的技术之一是使用FluxProcessor#Sink方法，如下例所示

@SpringBootApplication
@RestController
public class DemoApplication {

    final FluxProcessor processor;
    final FluxSink sink;
    final AtomicLong counter;

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);

    }

    public DemoApplication() {
        this.processor = DirectProcessor.create().serialize();
        this.sink = processor.sink();
        this.counter = new AtomicLong();
    }

    @GetMapping("/send")
    public void test() {
        sink.next("Hello World #" + counter.getAndIncrement());
    }

    @RequestMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent> sse() {
        return processor.map(e -> ServerSentEvent.builder(e).build());
    }
}

在这里，我创建了directprocessor以支持多个订阅服务器，这些订阅服务器将侦听数据流。此外，我还提供了额外的fluxprocessor#serialize，它为多生产者提供了安全的支持（从不同线程调用而不违反Reactive Streams规范规则，特别是规则1.3）。最后，通过调用“http：//localhost:8080/send”，我们将看到消息hello World#1（当然，只有在您之前连接到“http：//localhost:8080”的情况下才会出现）

在Reactor 3.4中，您有一个名为Reactor.core.publisher.sinks的新API。sinksAPI为手动数据发送提供了一个流畅的构建器，允许您指定数据流中的元素数量和背压行为、支持的订阅者数量和重播功能：

@SpringBootApplication
@RestController
public class DemoApplication {

    final Sinks.Many sink;
    final AtomicLong counter;

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);

    }

    public DemoApplication() {
        this.sink = Sinks.many().multicast().onBackpressureBuffer();
        this.counter = new AtomicLong();
    }

    @GetMapping("/send")
    public void test() {
        EmitResult result = sink.tryEmitNext("Hello World #" + counter.getAndIncrement());

        if (result.isFailure()) {
          // do something here, since emission failed
        }
    }

    @RequestMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent> sse() {
        return sink.asFlux().map(e -> ServerSentEvent.builder(e).build());
    }
}

注意，通过sinksAPI发送消息引入了emission及其结果的新概念。这种API的原因是反应器延伸反应物流，并且必须遵循背压控制。也就是说，如果发出的信号多于请求的信号，并且底层实现不支持缓冲，则不会传递消息。因此，TryEmitNext的结果返回EmitResult，该结果指示消息是否已发送。

另外，请注意，默认情况下sinskAPI给出了sink的序列化版本，这意味着您不必关心并发性。但是，如果事先知道消息的发射是串行的，则可以构建不序列化给定消息的sinks.unsafe()版本