如何让 Flux 的多个订阅者在不同的执行上下文/线程上运行

Question

我正在开发一个用于物联网实时数据可视化的 Spring Boot WebFlux 应用程序。

我有一个Flux，它模拟来自设备的数据，我希望在每个事件都建立 websocket 连接时：

• 必须通过 websocket 发送实时可视化（使用响应式WebSocketHandler）
• 必须根据给定条件进行检查，以便通过 HTTP REST 调用 (RestTemplate) 发送通知

从我的日志看来，两个订阅者（websocket 处理程序和通知程序）获得了两个具有完全不同值的不同流（在日志下方）。

我还尝试了在 MySource 类中的 map 之后链接 share 方法的变体，在这种情况下，看起来虽然我只有一个 Flux，但只有一个线程，所以一切都在阻塞（我可以看到 REST 调用阻止了通过 websocket 发送）。

这里发生了什么？我怎样才能让两个订阅者在不同的执行上下文（不同的线程）中运行，从而完全相互独立？

下面是相关代码sn-ps和logs。

提前谢谢大家！

更新： 为清楚起见，我必须指定 MyEvents 具有随机生成的值，因此由于 @NikolaB 的回答，我通过使用 ConnectableFlux / @ 解决了一个问题987654329@ 保证具有相同的Flux，但我仍然希望为两个订阅者提供单独的执行上下文。

public class MyWebSocketHandler implements WebSocketHandler {

   @Autowired
   public MySource mySource;

   @Autowired
   public Notifier notifier;

   public Mono<Void> handle(WebSocketSession webSocketSession) {
            Flux<MyEvent> events = mySource.events();
            events.subscribe(event -> notifier.sendNotification(event));
            return webSocketSession.send(events.map(this::toJson).map(webSocketSession::textMessage));
   }

   private String toJson(MyEvent event) {
       log.info("websocket toJson " + event.getValue());
       ...
   }
}

public class MySource {
   public Flux<MyEvent> events() {
      return Flux.interval(...).map(i -> new MyEvent(*Random Generate Value*);
   }
}

public class Notifier {

   public void sendNotification (MyEvent event) {
      log.info("notifier sendNotification " + event.getValue());
      if (condition met)
         restTemplate.exchange(...)
   }
}

2019-11-19 11:58:55.375 INFO [     parallel-3] i.a.m.websocket.MyWebSocketHandler  : websocket toJson 4.09
2019-11-19 11:58:55.375 INFO [     parallel-1] i.a.m.notifier.Notifier : notifier sendNotification 4.86
2019-11-19 11:58:57.366 INFO [     parallel-1] i.a.m.notifier.Notifier : notifier sendNotification 4.24
2019-11-19 11:58:57.374 INFO [     parallel-3] i.a.m.websocket.MyWebSocketHandler  : websocket toJson 4.11
2019-11-19 11:58:59.365 INFO [     parallel-1] i.a.m.notifier.Notifier : notifier sendNotification 4.61
2019-11-19 11:58:59.374 INFO [     parallel-3] i.a.m.websocket.MyWebSocketHandler  : websocket toJson 4.03
2019-11-19 11:59:01.365 INFO [     parallel-1] i.a.m.notifier.Notifier : notifier sendNotification 4.88
2019-11-19 11:59:01.375 INFO [     parallel-3] i.a.m.websocket.MyWebSocketHandler  : websocket toJson 4.29
2019-11-19 11:59:03.364 INFO [     parallel-1] i.a.m.notifier.Notifier : notifier sendNotification 4.37

Answer 1

这里有几个问题，首先RestTemplate 是同步/阻塞HTTP 客户端，所以你应该使用WebClient 这是反应式的，也可以创建ConnectableFlux（Flux 可以有多个订阅者）你需要在map 运算符之前分享它并创建新的Flux-es，它是从连接的一个创建的。

例子：

Flux<MyEvent> connectedFlux = mySource.events().share();
Flux.from(connectedFlux).subscribe(event -> notifier.sendNotification(event));
return webSocketSession.send(Flux.from(connectedFlux).map(this::toJson).map(webSocketSession::textMessage));

此外，sendNotification 方法应该返回 Mono<Void>，因为响应式方法应该总是返回 Mono 或 Fluxtypes。

要启动独立执行，您可以Zip 这两个Monos。

编辑

首先如上所述，将WebClient 用于传出HTTP 调用，这是响应式HTTP 客户端并重新编写Notifier 类：

public class Notifier {

   public Mono<Void> sendNotification (MyEvent event) {
      log.info("notifier sendNotification " + event.getValue());
      return Mono.just(event)
                 .filter(e -> /* your condition */)
                 .flatMap(e -> WebClient.builder().baseUrl("XXX")...)
                 .then();
   }

}

现在看看执行上下文是否不同。