Reactor 中错误是终止性事件，一旦 sink.emitError() 被调用，底层 Flux 就会永久终止，后续订阅均立即失败；解决关键是在消费者链路中隔离错误，而非在生产端（如 Redis 监听器）触发 emitError。
reactor 中错误是终止性事件，一旦 sink.emiterror() 被调用，底层 flux 就会永久终止，后续订阅均立即失败；解决关键是在**消费者链路中隔离错误**，而非在生产端（如 redis 监听器）触发 emiterror。
在使用 Sinks.Many 构建事件广播通道（例如 Redis 缓存更新推送）时，一个常见误区是：当业务逻辑抛出异常（如反序列化失败、校验异常），直接调用 sink.emitError() —— 这会导致整个 Flux 不可逆终止，所有现有及未来订阅者均收到 onError 并关闭，无法再接收新事件。
这与 Reactor 的 Reactive Streams 规范严格一致：错误是终端信号（terminal signal），不可恢复。因此，正确的设计原则是：
✅ 生产端（Sink）只负责“无副作用”地投递原始事件
❌ 绝不因单个事件处理失败而污染整个流生命周期
✅ 正确实践：错误隔离在消费链中
将潜在异常的处理逻辑下沉至 Flux 订阅链内部，使用 onErrorResume、onErrorReturn、handle() 等操作符局部兜底，确保单个事件失败不影响整体流持续运行。
以下是一个完整可运行示例，模拟多线程 Redis 消息监听场景：
private final Sinks.Many<string> sink = Sinks.many()
.multicast()
.onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, false);
// ✅ 生产端：仅安全 emitNext，完全避免 emitError
@Override
public void onMessage(Message message, byte[] pattern) {
String body = new String(message.getBody());
// 即使 body 为空或格式异常，也照常投递 —— 错误留给下游处理
sink.emitNext(body, Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(500)));
}
// ✅ 消费端：在 getPublisher() 返回的 Flux 中做健壮性处理
public Flux<string> getPublisher() {
return sink.asFlux()
.publishOn(Schedulers.single()) // 保持串行化处理（可按需调整）
.handle((event, sinkEmitter) -> {
try {
// 模拟业务处理：可能抛出 RuntimeException
if ("INVALID".equals(event)) {
throw new IllegalArgumentException("Invalid cache key");
}
sinkEmitter.next(processEvent(event)); // 如转换为 DTO、校验等
} catch (Exception e) {
// ⚠️ 关键：仅跳过当前事件，不中断流
log.warn("Skip invalid event: {}, cause: {}", event, e.getMessage());
// 可选：emit error 作为事件对象（如 ErrorEvent），而非流级错误
// sinkEmitter.next(new ErrorEvent(event, e));
}
})
.onErrorResume(e -> {
// 兜底：应对 handle 中未捕获的异常（极小概率）
log.error("Unexpected error in flux chain", e);
return Mono.empty(); // 继续空流，不终止
});
}
private String processEvent(String raw) {
// 示例：JSON 解析、业务转换等
return "PROCESSED_" + raw;
}</string></string>
? 订阅侧使用示例（带缓冲与容错）
redisPubSub.getPublisher()
.bufferTimeout(MAX_BUFFER_SIZE, MAX_BUFFER_DURATION)
.mapNotNull(batch -> {
try {
return transformAndPublish(batch); // 可能失败，但已被 handle 拦截
} catch (Exception e) {
log.error("Failed to process batch", e);
return null; // mapNotNull 自动过滤 null
}
})
.subscribe(
result -> log.info("Published: {}", result),
error -> log.error("UNEXPECTED TERMINAL ERROR — this should NOT happen!", error),
() -> log.info("Completed (won't occur unless sink is explicitly closed)")
);
⚠️ 重要注意事项
- 不要在 onMessage 中调用 emitError：Redis 监听器是多线程回调，emitError 会全局终止流，且无法重置 sink。
- Sinks.Many 不支持错误后恢复：它是单次终结型 sink；若需“重启流”，应重建 sink 实例（配合 Spring 生命周期管理）。
- 谨慎使用 retry() / repeat()：它们作用于整个 Flux，不适用于单事件错误；应优先用 handle() 或 flatMap(...onErrorResume...) 做细粒度控制。
- 日志与监控建议：对被跳过的事件记录 warn 日志，并考虑发送告警指标（如 invalid_event_count），便于问题追溯。
通过将错误处理下沉至消费链，你既能保障事件流的高可用性，又能精准定位和响应异常数据，真正实现“一个事件失败，不影响其他”的弹性设计目标。
