Spring Webfluxendpoint作为主题工作

Question

我有一个流量endpoint，我提供给客户机（订户），以接收更新的价格。我正在测试它通过浏览器访问URL(http://localhost:8080/prices)，它工作得很好。我面临的问题（我可能在这里遗漏了一些概念）是，当我在许多浏览器中打开这个URL，我希望在所有浏览器中都收到通知，但只有一个浏览器收到通知。它作为一个队列而不是一个主题工作（就像在message Brokers中一样）。那是正确的行为吗？

@GetMapping(value = "prices", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<Collection<Price>>> prices() {
return Flux.interval(Duration.ofSeconds(5))
        .map(sec -> pricesQueue.get())
        .filter(prices -> !prices.isEmpty())
        .map(prices -> ServerSentEvent.<Collection<Price>> builder()
                .event("status-changed")
                .data(prices)
                .build());
}

Answer 1

get不是标准的队列操作，但这几乎是肯定的，因为pricesqueue.get()方法不是幂等的。对于每个请求（在本例中打开的每个浏览器窗口），您将获得一个新的流量，该流量每5秒调用pricesqueue.get()。现在，如果pricesqueue.get()只检索队列中的最新项而不对其执行任何操作，那么一切都很好--所有订阅者都收到相同的项，并且显示相同的项。但是，如果它的行为更像poll()，它在检索到队列中的项后删除该项，那么只有第一个流量会得到该值--其余的流量不会，因为到那时它已经被删除了。

这里有两个主要的选择：

• 更改get()实现（或实现一个新方法），以便它不会改变队列，只检索一个值。
• 把焊剂变成热焊剂。将flux.interval(duration.ofseconds(5)).map(sec->pricesqueue.get()).publish().autoconnect()存储为字段（假设为queueflux)，然后在控制器方法中返回queueflux.filter(prices->！prices.isempty()).map(..)。