Flux.push通过在 的 lambda 表达式中调用和使用 sink直接创建 Fluxpush与使用 a 提供的 sink 有什么区别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，然后将事件发布到这个通量。
谁能告诉我，如果这两种方法都有效？当然，一定有一些不同。特别是，Reactor Reference Guide部分的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;
}
}
StateChangedEvents每当任何用户的状态发生变化时，非反应式业务逻辑都会发布：
@Data
@RequiredArgsConstructor
public class StateChangedEvent {
private final UUID userId;
private final ProcessState newState;
}
这就是我最初的问题的来源。我将如何构建一个将这个域事件转换为 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 事件确实会在从存储库中读取当前状态之前到达并发布，然后将其推送到无序状态。当然，在没有 synchronized 关键字的情况下，必须有一种更优雅的 Flux-ish 方式来处理这个问题。我们已经排除了
Flux.generate，它似乎可以使用Flux.push，Flux.create会产生更多的 SSE 事件吗？为什么？我害怕，我不明白这三个之间的区别。而不是使用静态方法，
Flux我应该在这里使用一个DirectProcessor或任何其他处理器吗？我是整个反应式堆栈的新手，Spring Reactor 文档对我来说太模糊了，tbh。再次：有什么区别？我上面提到的关于背压的评论呢？