如何在非反应性Spring EventListener和反应性Flux之间进行桥接

Question

通过调用Flux.push直接创建通量并在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和Spring WebFlux之间建立一个桥梁，在这里我想为每个传入的特定资源SSE请求创建一个Flux，然后将事件发布到这个Flux。

另一方面，它有不能处理背压的局限性。因此，如果您通过DirectProcessor推送N个元素，但至少有一个订阅服务器请求的元素少于N，DirectProcessor就会向它的订阅服务器发出IllegalStateException信号。

那是什么意思？

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

html" target="_blank">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;
}

这就是我最初的问题的来源。我如何建立一个桥梁，将这个领域的事件转换成一个通量流？我的要求：

• 新客户端注册后，应立即推送进程的当前状态
• 只要达到“终端”登机状态，通量流就应终止。

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

}

我的bridge代码中有几个问题我无法理解：

>

我真的必须同步我的subscriber实例，以避免从poll写入初始状态的过时事件吗？如果不这样做，就会发生StateChange事件到达并在从存储库读取当前状态之前发布，然后将其按顺序推出。当然，在没有synchronized关键字的情况下，必须有一种更优雅的Flux式方法来处理这个问题。

我们已经排除了flux.generate，它似乎可以使用flux.push，flux.create将生成更多的SSE事件？为什么？恐怕，我不明白这三者之间的区别。

与其使用flux上的静态方法，我应该使用directprocessor或这里的任何其他处理器吗？我对整个Reactor栈是新的，而且Spring Reactor文档对我来说太模糊了，TBH。再问一遍：区别在哪里？那我上面提到的关于背压的评论呢？

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

}

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