Spring WebFlux rest 控制器仅服务于前两个订阅

Question

有一个由 Flux.create 方法以编程方式创建的 flux：

Flux<Tweet> flux = Flux.<Tweet>create(emitter -> ...);

有一个rest控制器：

@RestController
public class StreamController {
    ...

    @GetMapping("/top-words")
    public Flux<TopWords> streamTopWords() {
        return topWordsStream.getTopWords();
    }
}

有几个web客户端(在独立的进程中):

Flux<TopWords> topWordsFlux = WebClient.create(".../top-words")
        .method(HttpMethod.GET)
        .accept(MediaType.TEXT_EVENT_STREAM)
        .retrieve()
        .bodyToFlux(TopWords.class)
        .subscribe(System.out::println);

JavaScript中有几个EventSource实例：

var eventSource = new EventSource(".../top-words");

eventSource.onmessage = function (e) {
    console.log("Processing message: ", e.data);
};

只有前两个“订户”将开始接收消息(不管它是web客户端还是EventSource实例)。另一个将打开连接，获取HTTP 200状态，但是事件流保持为空。客户端和服务器端都没有错误。

我不明白，对“2个订阅者”的限制在哪里。如果我想支持 2 个以上的订阅者，我该怎么办？

该应用程序使用Spring Boot 2.0.0.RELEASE构建，并使用Spring Boot starter webflux自动配置。默认配置不会更改。

Answer 1

我试图适应的基础API（Twitter流媒体API）中存在限制。

目标是连接到Twitter一次，并处理不同订阅者的推文流。

最初我认为传递给Flux.create方法的发射器总是对所有订阅者使用相同的FLuxSink。这当然没有意义。相反，正如javadoc明确指出的那样，FLuxSink是为每个订阅者提供的。

我使用Twitter侦听器实现了我的用例，该侦听器允许注册（和取消注册）许多FLuxSink实例。这样，单个推文流可以被各种不同的子用户订阅。

Flux<Tweet> flux = Flux.<Tweet>create(twitterListener::addSink);

我的twitterListener实现了org.springframework.social.twitter.api。StreamListener来自spring社交twitter项目。