基于Spring Boot RSocket的多租户聊天应用定向消息路由问题

你需要先维护所有在线用户的RSocket连接会话，针对多租户场景可以用线程安全的嵌套Map存储会话：

@Component
public class OnlineUserHolder {
    // 存储结构：租户ID -> (用户名 -> 对应用户的RSocket会话)
    private final ConcurrentMap<String, ConcurrentMap<String, RSocketSession>> tenantOnlineUsers = new ConcurrentHashMap<>();

    // 用户上线时添加会话
    public void addOnlineUser(String tenantId, String username, RSocketSession session) {
        ConcurrentMap<String, RSocketSession> userMap = tenantOnlineUsers.computeIfAbsent(tenantId, k -> new ConcurrentHashMap<>());
        userMap.put(username, session);
        // 监听会话关闭事件，用户下线时自动清理
        session.onClose().doFinally(signal -> {
            ConcurrentMap<String, RSocketSession> existMap = tenantOnlineUsers.get(tenantId);
            if (existMap != null) {
                existMap.remove(username);
                if (existMap.isEmpty()) {
                    tenantOnlineUsers.remove(tenantId);
                }
            }
        }).subscribe();
    }

    // 查询指定租户的在线用户列表
    public List<String> getTenantOnlineUsers(String tenantId) {
        ConcurrentMap<String, RSocketSession> userMap = tenantOnlineUsers.get(tenantId);
        return userMap == null ? Collections.emptyList() : new ArrayList<>(userMap.keySet());
    }

    // 获取指定租户下单个用户的会话
    public Optional<RSocketSession> getUserSession(String tenantId, String username) {
        ConcurrentMap<String, RSocketSession> userMap = tenantOnlineUsers.get(tenantId);
        return userMap == null ? Optional.empty() : Optional.ofNullable(userMap.get(username));
    }

    // 获取指定租户所有在线用户的会话
    public List<RSocketSession> getTenantAllSessions(String tenantId) {
        ConcurrentMap<String, RSocketSession> userMap = tenantOnlineUsers.get(tenantId);
        return userMap == null ? Collections.emptyList() : new ArrayList<>(userMap.values());
    }
}

然后通过@ConnectMapping监听RSocket连接建立事件，存入用户会话：

@Controller
public class RSocketController {
    @Resource
    private OnlineUserHolder onlineUserHolder;

    @ConnectMapping
    public Mono<Void> onConnect(RSocketSession session, @AuthenticationPrincipal UserDetails user) {
        // 需在你的UserDetails实现类中扩展tenantId字段，存储用户所属租户
        String tenantId = ((CustomUserDetails) user).getTenantId();
        onlineUserHolder.addOnlineUser(tenantId, user.getUsername(), session);
        return Mono.empty();
    }

    // 提供查询在线用户的接口
    @MessageMapping("user.online.list")
    public Mono<List<String>> getOnlineUserList(@AuthenticationPrincipal UserDetails user) {
        String tenantId = ((CustomUserDetails) user).getTenantId();
        return Mono.just(onlineUserHolder.getTenantOnlineUsers(tenantId));
    }
}

首先扩展你的Message实体，新增必要字段：租户ID、接收人标识（群发填ALL，私信填目标用户名）、发送时间、是否已读等。
然后修改消息接收逻辑，完成消息存储+路由分发：

@Resource
private MessageRepository messageRepository; // 自行实现消息持久化的Repository

@MessageMapping("message.send")
public Mono<Void> receiveAndRouteMessage(Message message, @AuthenticationPrincipal UserDetails user) {
    // 补全消息基础属性
    String tenantId = ((CustomUserDetails) user).getTenantId();
    message.setTenantId(tenantId);
    message.setSender(user.getUsername());
    message.setSendTime(LocalDateTime.now());

    // 1. 先持久化消息到数据库
    return messageRepository.save(message)
            .then(Mono.defer(() -> {
                // 2. 判断是群发还是私信
                if ("ALL".equals(message.getReceiver())) {
                    // 租户群发：给租户所有在线用户推送消息
                    List<RSocketSession> sessions = onlineUserHolder.getTenantAllSessions(tenantId);
                    return Flux.fromIterable(sessions)
                            .flatMap(session -> session.send(
                                    Mono.just(session.payloadEncoder().encode(
                                            DefaultPayload.create(new ObjectMapper().writeValueAsBytes(message)),
                                            session.metadataMimeType()
                                    ))
                            )).then();
                } else {
                    // 私信：找到目标用户会话推送
                    return onlineUserHolder.getUserSession(tenantId, message.getReceiver())
                            .map(session -> session.send(
                                    Mono.just(session.payloadEncoder().encode(
                                            DefaultPayload.create(new ObjectMapper().writeValueAsBytes(message)),
                                            session.metadataMimeType()
                                    ))
                            )).orElse(Mono.empty());
                }
            }));
}

React端使用rsocket-js库建立连接后，订阅接收消息的路由即可收到服务端推送的消息：

// 连接建立完成后订阅消息推送
requester.requestStream({
  metadata: String.fromCharCode('message.receive'.length) + 'message.receive'
}).subscribe({
  onNext: (payload) => {
    const message = JSON.parse(payload.data);
    // 自行处理消息渲染逻辑
  }
})

• 单实例部署时上述内存会话存储完全可用，多实例部署时需要将在线用户信息存入Redis，配合跨实例消息广播机制实现全量推送
• 认证逻辑需要确保用户的租户ID、用户名不可篡改，避免跨租户消息泄露

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