SpringBoot中如何管理多个RSocket客户端

你猜想的用Map存储每个客户端的RSocketRequester是完全正确的思路，这是Spring环境下实现服务器主动向多客户端发起请求的标准方案，具体实现可以按以下步骤来：

1. 客户端连接时注册RSocketRequester #

服务器需要在客户端发起连接时，捕获并存储每个客户端的连接实例：

• 用@ConnectMapping注解定义连接处理方法，客户端连接到服务器时会触发该方法；
• 要求客户端在连接时携带唯一标识（比如客户端ID），可以通过元数据或连接Payload传递；
• 用线程安全的Map（比如ConcurrentHashMap）存储RSocketRequester，以客户端ID作为key；
• 绑定连接关闭回调，客户端断开时自动从Map中移除对应条目，避免内存泄漏。

示例代码：

@Controller
public class ClientConnectionController {
    private final Map<String, RSocketRequester> clientRequesters = new ConcurrentHashMap<>();

    @ConnectMapping("client-connect")
    public void handleClientConnect(RSocketRequester requester, @Payload String clientId) {
        // 配置Requester保持连接并监听关闭事件
        requester = requester.rsocket()
                .onClose()
                .doFirst(() -> clientRequesters.put(clientId, requester))
                .doOnError(error -> clientRequesters.remove(clientId))
                .doFinally(signal -> clientRequesters.remove(clientId))
                .subscribe();
    }
}

2. 主动向指定客户端发送请求 #

从Map中取出目标客户端的RSocketRequester，调用requestResponse()方法发起请求并获取响应：

• 用route()指定客户端侧的@MessageMapping路由；
• 用data()传递请求数据；
• 通过retrieveMono()或retrieveFlux()获取客户端的响应结果。

示例代码：

@Service
public class ClientRequestService {
    private final Map<String, RSocketRequester> clientRequesters;

    public ClientRequestService(Map<String, RSocketRequester> clientRequesters) {
        this.clientRequesters = clientRequesters;
    }

    public Mono<ClientResponse> sendRequestToClient(String clientId, ClientRequest request) {
        RSocketRequester requester = clientRequesters.get(clientId);
        if (requester == null) {
            return Mono.error(new RuntimeException("客户端未连接: " + clientId));
        }
        return requester.route("client-process-request")
                .data(request)
                .retrieveMono(ClientResponse.class);
    }
}

3. 关键注意事项 #

• 线程安全：必须使用线程安全的Map实现，避免多线程操作时出现并发问题；
• 连接生命周期管理：务必处理连接关闭事件，及时清理无效的RSocketRequester实例；
• 客户端标识唯一性：确保每个客户端的ID唯一，避免Map中出现覆盖或混淆；
• 异常处理：发送请求时要处理客户端不存在、连接失效等异常情况，避免服务崩溃。

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