关于Flux Spring Events未送达消息重发机制的文档咨询

核心背景与疑问 #

• 配置Event Source时，通过背压缓冲区暂存待发送给监听器的消息：

  Sinks.many()
       .multicast()
       .onBackpressureBuffer(1000)
  

• 可通过ServerSentEvent为每条消息绑定唯一ID：

  ServerSentEvent.builder(e.data).id(String.valueOf(e.id)).build()
  

• 核心疑问：前端携带lastEventId重连时，仅依赖backpressure buffer无法保证从指定ID续接消息——backpressure buffer仅用于临时暂存因背压无法发送的消息，一旦缓冲区填满就会丢弃新消息，且没有按ID索引、持久化消息的能力，无法支撑断点续传需求。

官方文档相关内容参考 #

1. Server-Sent Events 断点续传的本质 #

Spring官方文档明确说明，lastEventId的作用是告知服务器客户端最后成功接收的消息ID，但默认的backpressure buffer并不具备消息回溯能力。要实现从指定ID开始推送消息，必须自行维护一个具备以下特性的消息存储：

• 支持按ID索引查询后续消息
• 配置最大容量（避免内存溢出）
• 设置过期时间（自动清理无效旧消息）

2. Sinks 缓冲区的选型边界 #

官方文档指出，onBackpressureBuffer仅适用于处理瞬时背压场景的临时消息暂存，完全不适用于需要断点续传的业务场景。若要实现自定义断点续传，需替换为以下方案：

• 使用带过期策略的内存存储（如Guava Cache）存储消息，并关联消息ID
• 重写Sinks的消息发送逻辑，将消息同步写入自定义缓冲区
• 处理客户端重连请求时，根据lastEventId从缓冲区中筛选并推送后续消息

3. 代码实践指引 #

实现自定义断点续传的核心步骤：

• 初始化带容量和过期限制的消息缓存：

  LoadingCache<Long, EventData> messageCache = CacheBuilder.newBuilder()
      .maximumSize(5000)
      .expireAfterWrite(1, TimeUnit.HOURS)
      .build(key -> /* 缓存无数据时，可从持久化存储补充 */);
  

• 发送消息时同步写入缓存：

  sink.emitNext(ServerSentEvent.builder(event.getData()).id(String.valueOf(event.getId())).build(), Sinks.EmitFailureHandler.FAIL_FAST);
  messageCache.put(event.getId(), event.getData());
  

• 处理重连请求时根据lastEventId推送后续消息：

  String lastEventId = request.headers().firstHeader("Last-Event-ID");
  long startId = lastEventId != null ? Long.parseLong(lastEventId) + 1 : 0;
  return Flux.fromStream(messageCache.asMap().entrySet().stream()
      .filter(entry -> entry.getKey() >= startId)
      .map(entry -> ServerSentEvent.builder(entry.getValue()).id(String.valueOf(entry.getKey())).build())
      .sorted(Map.Entry.comparingByKey()))
      .concatWith(sink.asFlux());
  

结论 #

默认的backpressure buffer仅能处理瞬时背压场景，无法满足前端从指定ID开始接收消息的需求。必须通过自定义带容量和过期策略的消息缓冲区，配合lastEventId参数，才能实现可靠的断点续传功能。

内容的提问来源于stack exchange，提问作者Chris Cleverley