RSocket请求流完成后如何监听新事件？

当前基于Spring RSocket实现聊天消息推送，后端端点与前端客户端代码如下，遇到的问题是：requestStream传输完现有消息后触发complete回调，后续新产生的聊天消息无法被客户端接收。目前用轮询方式规避，但会导致重复接收消息，希望得到更优方案。

后端RSocket端点代码 #

@MessageMapping("chat.{chatId}")
Flux<Message> getChats(@DestinationVariable String chatId) {
    Mono<Chat> data = chatRepository.findById(chatId);
    return data.map(chatGroup -> chatGroup.getMessages())
            .flatMapMany(Flux::fromIterable);
}

public interface ChatRepository extends FirestoreReactiveRepository<Chat> {

}

前端RSocket客户端代码（RSocket 1.x） #

const makeConnector = () => {
      return new RSocketConnector({
        setup: {
          dataMimeType: 'application/json',
          keepAlive: 100000,
          lifetime: 100000,
          metadataMimeType: 'message/x.rsocket.routing.v0',
        },
        transport: new WebsocketClientTransport({
          url: 'ws://localhost:7000/rsocket',
        }),
      });
};

client()
  .connect()
  .then((socket) => {
    const requester = socket.requestStream(
      {
        data: undefined,
        metadata: Buffer.concat([
          Buffer.from(String.fromCharCode('chat.chatId'.length)),
          Buffer.from('chat.chatId'),
        ]),
      },
      10,
      {
        onError: (e) => console.log('error getting data', e),
        onNext: (payload, isComplete) => {
          const parsedData = JSON.parse(payload.data.toString());
          requester.request(5); // 请求下5条消息
        },
        onComplete: () => {
          console.log('complete');
        },
        onExtension: () => {
          console.log('on extension');
        },
      }
    );
  });

当前轮询规避方案（存在重复接收问题） #

return Flux.interval(Duration.ofSeconds(5)).flatMap(x -> {
   return chatService.findChats(matchId);
});

问题核心：当前后端返回的是一次性有限流——从Chat对象的messages列表转换而来，所有现有消息推送完成后流直接终止，自然无法感知后续新消息。要实现持续监听，需将端点改为无限流，结合数据库实时监听或事件订阅机制。

后端修改方案 #

方案1：基于Firestore实时变更监听 #

利用Firestore的文档变更监听能力，实现消息的实时推送：

@Autowired
FirebaseFirestore firestore;

@MessageMapping("chat.{chatId}")
Flux<Message> getChats(@DestinationVariable String chatId) {
    // 1. 加载初始消息
    Mono<List<Message>> initialMessages = firestore.collection("chats")
            .document(chatId)
            .get()
            .map(doc -> doc.toObject(Chat.class).getMessages());
    
    // 2. 实时监听文档变更，推送新消息
    Flux<List<Message>> updatedMessages = Flux.create(sink -> {
        ListenerRegistration registration = firestore.collection("chats")
                .document(chatId)
                .addSnapshotListener((snapshot, error) -> {
                    if (error != null) {
                        sink.error(error);
                        return;
                    }
                    if (snapshot != null && snapshot.exists()) {
                        Chat chat = snapshot.toObject(Chat.class);
                        sink.next(chat.getMessages());
                    }
                });
        // 客户端断开时移除监听，避免资源泄漏
        sink.onCancel(() -> registration.remove());
    });
    
    // 3. 合并初始消息与变更消息，通过消息ID去重
    return initialMessages.flatMapMany(Flux::fromIterable)
            .concatWith(updatedMessages.flatMap(Flux::fromIterable))
            .distinct(message -> message.getId());
}

方案2：基于Spring事件发布订阅 #

若无需数据库层面监听，可自定义事件机制实现新消息推送：

1. 定义新消息事件类

public class NewMessageEvent {
    private String chatId;
    private Message message;

    public NewMessageEvent(String chatId, Message message) {
        this.chatId = chatId;
        this.message = message;
    }

    // getter方法
    public String getChatId() { return chatId; }
    public Message getMessage() { return message; }
}

2. 新增消息时发布事件

@Autowired
ApplicationEventPublisher eventPublisher;

public void addMessage(String chatId, Message message) {
    // 保存消息到数据库
    chatRepository.findById(chatId)
            .map(chat -> {
                chat.getMessages().add(message);
                return chat;
            })
            .flatMap(chatRepository::save)
            .subscribe();
    // 发布新消息事件
    eventPublisher.publishEvent(new NewMessageEvent(chatId, message));
}

3. 修改RSocket端点，合并初始消息与事件流

@Autowired
ApplicationEventPublisher publisher;

@MessageMapping("chat.{chatId}")
Flux<Message> getChats(@DestinationVariable String chatId) {
    // 初始消息流
    Flux<Message> initialMessages = chatRepository.findById(chatId)
            .flatMapMany(chat -> Flux.fromIterable(chat.getMessages()));
    
    // 新消息事件流，仅接收当前聊天ID的事件
    Flux<Message> newMessageStream = Flux.create(sink -> {
        ApplicationListener<NewMessageEvent> listener = event -> {
            if (event.getChatId().equals(chatId)) {
                sink.next(event.getMessage());
            }
        };
        publisher.addApplicationListener(listener);
        // 订阅取消时移除监听器
        sink.onCancel(() -> publisher.removeApplicationListener(listener));
    });
    
    // 合并流：先推初始消息，再持续推送新消息
    return initialMessages.concatWith(newMessageStream);
}

客户端调整 #

后端改为无限流后，流不会自动终止，客户端只需保持订阅即可。若需背压控制，可根据自身处理能力按需请求更多消息：

client()
  .connect()
  .then((socket) => {
    const targetChatId = "实际聊天ID";
    const requester = socket.requestStream(
      {
        data: undefined,
        metadata: Buffer.concat([
          Buffer.from(String.fromCharCode(`chat.${targetChatId}`.length)),
          Buffer.from(`chat.${targetChatId}`),
        ]),
      },
      10, // 初始请求10条消息
      {
        onError: (e) => console.log('获取数据出错', e),
        onNext: (payload, isComplete) => {
          const parsedData = JSON.parse(payload.data.toString());
          console.log('收到消息', parsedData);
          // 按需请求更多消息（背压控制）
          // if (本地消息缓存剩余空间 > 0) {
          //   requester.request(5);
          // }
        },
        onComplete: () => {
          // 正常无限流不会触发此回调，除非后端主动终止
          console.log('流已完成');
        },
        onExtension: () => {
          console.log('收到扩展');
        },
      }
    );
  });

关键注意事项 #

• 资源清理：后端需在客户端断开时移除监听/取消订阅，避免内存泄漏。
• 去重处理：使用数据库变更监听时，每次变更会返回完整消息列表，需通过消息ID去重避免重复推送。
• 背压控制：根据客户端处理能力调整请求量，防止消息堆积导致内存溢出。

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