如何在非反应性 Spring EventListener 和反应性 Flux 之间架起桥梁

Question

通过调用 Flux.push 直接创建 Flux 并使用 push 的 lambada 表达式中的接收器与使用 DirectProcessor 提供的接收器有什么区别？

在一个 Flux 仅发出几个事件的最小示例中，我可以这样做

Flux.<String>push(emitter -> {
   emitter.next("One");
   emitter.next("Two");
   emitter.complete();
 });

对比使用 DirectProcessor

var emitter = DirectProcessor.<String>create().sink();
emitter.next("One");
emitter.next("Two");
emitter.complete();

澄清一下：我知道我可以在这里使用 Flux.just，但我的用例实际上是在 Spring 的 @EventListener 和 [=98= 之间架起一座桥梁] WebFlux，我想为特定资源的每个传入 SSE 请求创建一个 Flux，然后将事件发布到这个 Flux。

谁能告诉我这两种方法是否都有效？当然，必须有一些区别。特别是 Reactor Reference Guide 关于 DirectProcessor 的部分指出：

On the other hand, it has the limitation of not handling backpressure. As a consequence, a DirectProcessor signals an IllegalStateException to its subscribers if you push N elements through it but at least one of its subscribers has requested less than N.

这是什么意思？

[EDIT:] 在问题的早期版本中我使用 Flux.generate() 而不是 Flux.push()，这显然是错误的，因为生成可以最多创建一个活动。

[编辑 2:] @123 要求我提供一个完整的例子来说明我正在努力实现的目标。请耐心等待，这是一个 SO 问题的相当多的代码：

我实际尝试做的事情的完整示例

我想在（非反应性）Spring 域事件侦听器和反应性 Flux 之间建立一座桥梁，然后我可以在 WebFlux 端点中使用它来发布 SSE。为简洁起见，以下代码片段使用 Lombok 注释。

假设我最终想在入职流程中将用户状态发布为 SSE。这是枚举：

public enum ProcessState {
  CREATED(false),
  VERIFIED(false),
  AUTHORIZATION_PENDING(false),
  AUTHORIZED(false),
  ACTIVE(true);

  @Getter
  private final boolean terminalState;

  ProcessState(boolean terminalState) {
    this.terminalState = terminalState;
  }

}

只要任何用户的状态发生变化，非反应性业务逻辑就会发布 StateChangedEvents：

@Data
@RequiredArgsConstructor
public class StateChangedEvent {
  private final UUID userId;
  private final ProcessState newState;
}

这就是我最初的问题的来源。我将如何构建将此域事件转换为 Flux 流的桥梁？我的要求：

进程的当前状态应在新客户端注册后立即推送
只要达到 "terminal" 入职状态，Flux 流就应该终止。

这是我目前得到的：

@Component
@RequiredArgsConstructor
class EventBridge {

  @RequiredArgsConstructor(access = PRIVATE)
  private static class Subscriber {
    private final UUID userId;
    private final FluxSink<ProcessState> sink;
    private boolean eventEmitted;
  }

  private final UserRepository repository;
  private final Map<UUID, Subscriber> subscribers = new ConcurrentHashMap<>();

  @EventListener
  void stateChanged(StateChangedEvent event) {
    notifySubscribers(event);
  }

  Flux<ProcessState> register(UUID userId) {
    return Flux.push(emitter -> addSubscriber(userId, emitter));
  }

  private Subscriber addSubscriber(UUID userId, FluxSink<ProcessState> sink) {
    var subscriptionId = randomUUID();
    var subscriber = new Subscriber(userId, sink);
    subscribers.put(subscriptionId, subscriber);
    sink
      .onRequest(n -> poll(subscriber))
      .onDispose(() -> removeSubscriber(subscriptionId));
    return subscriber;
  }

  private void poll(Subscriber subscriber) {
    emit(subscriber, loadCurrentState(subscriber), true);
  }

  private ProcessState loadCurrentState(Subscriber subscriber) {
    return repository.findById(subscriber.userId).getProcessState();
  }

  private void removeSubscriber(UUID subscriptionId) {
    subscribers.remove(subscriptionId);
  }

  private void notifySubscribers(StateChangedEvent event) {
    subscribers.values().stream()
      .filter(subscriber -> subscriber.userId.equals(event.getUserId()))
      .forEach(subscriber -> emit(subscriber, event.getNewState(), false));
  }

  private void emit(Subscriber subscriber, ProcessState processState, boolean onlyIfFirst) {
    synchronized (subscriber) {
      if (onlyIfFirst && subscriber.eventEmitted) {
        return;
      }
      subscriber.sink.next(processState);
      if (processState.isTerminalState()) {
        subscriber.sink.complete();
      }
      subscriber.eventEmitted = true;
    }
  }

}

最后是使用网桥的控制器：

@RestController
@RequiredArgsConstructor
class UserController {

  private final EventBridge eventBridge;

  @GetMapping(value = "/{userId}", produces = TEXT_EVENT_STREAM_VALUE)
  Flux<ServerSentEvent<ProcessState>> readAsStream(@PathVariable UUID userId) {
    return eventBridge.register(userId).map(response -> ServerSentEvent.builder((ProcessState) response).build());
  }

}

我的桥接代码中有几个问题我无法解决：

我真的必须在我的 Subscriber 实例上同步以避免从 poll 初始状态写入陈旧事件吗？如果我不这样做，确实会发生 StateChange 事件到达并发布 before 从存储库读取当前状态，即 then 推送出问题了。当然，必须有一种更优雅的 Flux-ish 方式来处理这个问题，而不需要同步关键字。
我们已经排除了Flux.generate，似乎与Flux.push一起工作，Flux.create会产生更多的SSE事件？为什么？恐怕我不明白这三者的区别
而不是在 Flux 上使用静态方法，我应该在这里使用 DirectProcessor 还是任何其他处理器？我是整个反应堆的新手， Spring Reactor 文档对我来说太模糊了，老实说。再次：有什么区别？我上面提到的关于背压的评论怎么样？

Answer 1

因此，如果我理解您正在尝试正确执行的操作，我认为您的解决方案可以大大简化。

@Component
public class EventBridge {

    private final UserRepository repository;
    private final ReplayProcessor<StateChangedEvent> processor;
    private final FluxSink<StateChangedEvent> sink;


    EventBridge(UserRepository repository){
        this.repository= repository;
        //Replays events from last 100S for every new subscriber
        this.processor = ReplayProcessor.createTimeout(Duration.ofSeconds(100L));
        //Sink provides thread safe next,complete and error for subscribers
        this.sink = processor.sink();
    }

    public void changeState(StateChangedEvent event) {
        //Literally just pass event into sink, calls onNext on subscribers
        sink.next(event);
    }

    public Flux<ProcessState> streamProcessStateForUser(UUID uuid){
        return
                //Search repository first, this isn't great since it blocks until 
                //the repo returns, although that seems to be what you want
                //Also I added an Unknown to ProcessState, since it's better than 
                //it being null. 
                //Also you should probably return optional from repo. 
            Flux.concat(
                    Flux.just(
                            userRepo.findById(uuid).map(User::getProcessState).orElse(ProcessState.UNKNOWN)
                    ),
                    processor
                            //Check the uuid matches the event
                            .filter(stateChangedEvent -> stateChangedEvent.getUserId().equals(uuid))
                            //Time out after 100 seconds, not needed but may be useful for you
                            .take(Duration.ofSeconds(100L))
                            //Complete flux when at terminal state
                            .takeUntil(stateChangedEvent -> stateChangedEvent.getNewState().isTerminalState())
                            //Convert to ProcessState from StateChangedEvent
                            .map(StateChangedEvent::getNewState)
            );
    }

}

应该能够保持其他一切不变。

Answer 2

感谢@123 回答我关于如何在 Spring 的 @EventListener 和 Flux 之间架起桥梁的问题。如问题中所述，完整的用例是使用 WebFlux 的 SSE 支持将域事件推送到客户端。

经过一番思考，我意识到一开始就构建这个桥没有意义，因为在多实例场景中，HTTP 请求可能会到达与请求所在的实例不同的实例。入职流程是运行，因此根本不会推送任何事件。

所以最后，我选择轮询单一事实来源——数据库——并在状态发生变化时推送 SSE 事件。如果我们可以在这里使用反应式数据存储，那就太好了，但现在我 "stuck" 使用 Spring Data JPA 和 PostgreSQL。

所以，如果有人有同样的问题，这就是我最终构建它的方式：

@RestController
@RequiredArgsConstructor
class UserController {

  private final UserRepository userRepository;

  @GetMapping(value = "/{userId}", produces = TEXT_EVENT_STREAM_VALUE)
  Flux<ServerSentEvent<ProcessState>> readAsStream(@PathVariable UUID userId) {
    return Flux.interval(ZERO, ofSeconds(1L))
      .map(n -> userRepository.findById(userId).getProcessState())
      .takeUntil(processState -> processState.isTerminalState())
      .distinctUntilChanged()
      .map(response -> ServerSentEvent.builder((ProcessState) response).build())
      .take(ofMinutes(30L));
  }

}

以防万一有人想知道：这再次被简化以说明手头的问题。我们有一个六边形架构，即不要在我们的 @RestController 中注入 Repositories，而是从我们的 Web 层调用业务外观，也就是输入端口来检索用户。

如何在非反应性 Spring EventListener 和反应性 Flux 之间架起桥梁

How to bridge between a non-reactive Spring EventListener and a reactive Flux

spring

spring-webflux

spring-reactive

spring-reactor

我实际尝试做的事情的完整示例