如何基于Spring Boot RSocket与Angular rsocket-js实现服务端主动推送通知

核心逻辑说明 #

RSocket连接是全双向的，之前你只用到了「客户端发起请求-服务端响应」的路径，只要在服务端持有每个客户端连接的会话实例，就可以主动发起Fire-and-Forget类型的消息推送，不需要依赖客户端的请求触发。

后端改造步骤 #

1. 实现全局RSocket会话管理器 #

专门用来存储所有在线客户端的连接实例，提供批量/定向推送能力，代码如下：

@Component
public class RSocketSessionManager {
    // 线程安全的客户端连接集合
    private final Set<RSocketRequester> onlineClients = ConcurrentHashMap.newKeySet();

    // 新增客户端连接
    public void registerClient(RSocketRequester requester) {
        onlineClients.add(requester);
        // 监听连接断开事件，自动移除无效会话
        requester.rsocket().onClose()
                 .doFinally(signal -> onlineClients.remove(requester))
                 .subscribe();
    }

    // Fire-and-Forget方式推送通知给所有在线客户端
    public void broadcastNewNotification(Notification notification) {
        onlineClients.forEach(client -> {
            // 路由到客户端的消息处理端点，send()对应RSocket的Fire-and-Forget语义
            client.route("client.push.notification")
                  .data(notification)
                  .send()
                  .subscribe();
        });
    }
}

2. 捕获客户端连接事件 #

通过@ConnectMapping注解的方法，在客户端完成RSocket握手时自动注册会话：

@Controller
public class RSocketConnectController {
    private final RSocketSessionManager sessionManager;

    public RSocketConnectController(RSocketSessionManager sessionManager) {
        this.sessionManager = sessionManager;
    }

    // 客户端握手完成后触发，注入当前连接对应的RSocketRequester
    @ConnectMapping
    public void onClientHandshakeComplete(RSocketRequester requester) {
        sessionManager.registerClient(requester);
    }

    // 你已实现的历史通知拉取接口完全保留，不需要改动
    @MessageMapping("streams")
    public Flux<Notification> requestStream() {
        log.info("Streaming to notifications...");
        return streamEventService.retrieveAllNotifications().log();
    }
}

3. 业务触发推送逻辑 #

当数据库新增通知完成后，直接调用会话管理器的广播方法即可：

@Service
public class NotificationBizService {
    private final RSocketSessionManager sessionManager;
    private final NotificationRepository notificationRepository;

    public NotificationBizService(RSocketSessionManager sessionManager, NotificationRepository notificationRepository) {
        this.sessionManager = sessionManager;
        this.notificationRepository = notificationRepository;
    }

    public Notification saveNotification(Notification notification) {
        // 先落库
        Notification saved = notificationRepository.save(notification);
        // 落库完成后主动推送所有在线客户端
        sessionManager.broadcastNewNotification(saved);
        return saved;
    }
}

前端改造步骤 #

之前你用的EchoResponder是默认回声实现，无法处理后端主动推送的消息，需要替换为自定义的消息响应器：

1. 导入RSocket核心依赖，实现支持Fire-and-Forget消息处理的Responder
2. 把原代码中硬编码的路由字符串拼接替换为官方的路由编码方法，避免元数据格式错误
   修改后的前端服务代码如下：

import { RSocketClient, JsonSerializer, IdentitySerializer, Responder, encodeRoute } from 'rsocket-core';
import RSocketWebSocketClient from 'rsocket-websocket-client';
import { Subject } from 'rxjs';

// 自定义消息响应器，处理后端主动推来的Fire-and-Forget消息
class PushMessageResponder implements Responder {
  // 实现Fire-and-Forget的处理逻辑
  handleFireAndForget(payload: any) {
    const newNotification = payload.data;
    console.log('收到后端推送的新通知', newNotification);
    // 可通过Subject把消息传递给Angular组件更新UI
    return {
      onSubscribe: (subscription) => subscription.cancel()
    };
  }

  // 其他未使用的交互方法空实现即可
  handleRequestResponse() { return { onSubscribe: (sub) => sub.cancel() }; }
  handleRequestStream() { return { onSubscribe: (sub) => sub.cancel() }; }
  handleRequestChannel() { return { onSubscribe: (sub) => sub.cancel() }; }
}

export class RsocketClientService {
  private readonly wsURL = 'ws://localhost:7000/notification';
  client: any;
  socket: any;
  // 暴露推送事件给页面组件
  public newNotificationEvent = new Subject<Notification>();

  constructor() { 
    this.client = new RSocketClient({
      serializers: { data: JsonSerializer, metadata: IdentitySerializer },
      setup: {
        keepAlive: 10000,
        lifetime: 180000,
        dataMimeType: 'application/json',
        metadataMimeType: 'message/x.rsocket.routing.v0',       
        payload: { data: 23 }
      },
      transport: new RSocketWebSocketClient({ url: this.wsURL }),
      // 替换原EchoResponder为自定义实现
      responder: new PushMessageResponder()
    });
  }

  public connect() {
    this.client.connect().subscribe({
      onComplete: (socket: any) => {
        this.socket = socket;
        this.socket.connectionStatus().subscribe(status => console.log("连接状态更新", status));
      },
      onError: error => console.error("连接失败", error)
    });
  }

  public retrieveNotifications() {
    this.socket.requestStream({
      data: null,
      // 用官方方法编码路由，替代原来的手动字符串拼接
      metadata: encodeRoute('streams')
    }).subscribe({
      onNext: payload => console.log("拉取到历史通知", payload)
    });
  }
}

补充说明 #

• 如果需要实现定向推送（只给指定用户发通知），只要在registerClient时把用户ID和RSocketRequester的映射关系存入线程安全的Map，推送时按用户ID筛选即可，不需要遍历全部客户端
• 后端调用的send()方法严格遵循RSocket规范的Fire-and-Forget语义，发出消息后不需要等待客户端返回任何响应，完全匹配需求
• 注意保持前后端路由字符串的一致性，避免因为路由不匹配导致消息收不到

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