0

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

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

有一个休息控制器：

@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 自动配置。默认配置没有改变。

4

1 回答 1

0

我尝试适应的底层 API 存在限制（Twitter 流 API）。

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

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

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

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

我在 spring-social-twitter 项目中的twitterListener实现。org.springframework.social.twitter.api.StreamListener

于 2018-04-03T11:08:29.137 回答