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

Question

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

Ste*_*erl 5 spring spring-webflux spring-reactive spring-reactor

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

在 Flux 只发出几个事件的最小示例中，我可以做

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

Run Code Online (Sandbox Code Playgroud)

与使用 DirectProcessor

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

Run Code Online (Sandbox Code Playgroud)

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

有人能告诉我，这两种方法是否都有效？当然，肯定有一些不同。特别是，Reactor Reference Guide部分关于DirectProcessor状态：

另一方面，它具有不处理背压的限制。因此，如果您通过 DirectProcessor 推送 N 个元素但至少其中一个订阅者的请求少于 N，则 DirectProcessor 会向其订阅者发出 IllegalStateException 信号。

这意味着什么？

[编辑：]在我使用的问题的早期版本中，Flux.generate()而不是Flux.push()，这显然是错误的，因为 generate 最多可以创建一个事件。

[编辑 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;
  }

}

Run Code Online (Sandbox Code Playgroud)

StateChangedEvents每当任何用户的状态发生更改时，非反应式业务逻辑都会发布：

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

Run Code Online (Sandbox Code Playgroud)

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

新客户端注册后应立即推送进程的当前状态
每当达到“终端”载入状态时，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;
    }
  }

}

Run Code Online (Sandbox Code Playgroud)

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

@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());
  }

}

Run Code Online (Sandbox Code Playgroud)

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

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

Answer 1

123*_*123 6

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

@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)
            );
    }

}

Run Code Online (Sandbox Code Playgroud)

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

Answer 2

Ste*_*erl 5

感谢 @123 回答我关于如何在 Spring@EventListener和Flux. 正如问题中提到的，完整的用例是使用 WebFlux 的 SSE 支持将域事件推送到客户端。

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

所以最后，我选择轮询单一事实来源 - 数据库 - 并在状态发生变化时推送 SSE 事件。如果我们可以在这里使用反应式数据存储，那就太好了，但现在我“卡在”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));
  }

}

Run Code Online (Sandbox Code Playgroud)

以防万一有人想知道：这再次被简化以说明当前的问题。我们有一个六角形架构，即不注入Repositories我们的@RestControllers，而是从我们的 Web 层调用业务门面（又名输入端口）来检索用户。

归档时间：	5 年，7 月前
查看次数：	1861 次
最近记录：	5 年，6 月前