Spring Boot RSocket如何检测连接断开？断开时如何获取用户信息

RSocket完全可以检测到用户断开连接，你当前的代码已经通过reqer.rsocket().onClose()监听了连接关闭事件，但缺少用户信息与RSocket连接的绑定逻辑，只需调整存储结构并补充关联逻辑即可实现需求。

1. 调整存储结构，绑定用户信息与连接实例 #

你当前仅用List<RSocketRequester>存储连接实例，无法关联用户身份信息。建议自定义实体类或使用Map，将用户信息与对应的RSocket连接绑定：

方式一：自定义连接实体类 #

// 自定义实体，封装用户信息和RSocket连接
@Data
public class ClientConnection {
    private String userId; // 用户唯一标识
    private String username; // 用户名
    private RSocketRequester requester; // 对应的RSocket连接实例
}

// 替换原有列表，使用线程安全集合避免并发问题
private final List<ClientConnection> CLIENT_CONNECTIONS = new CopyOnWriteArrayList<>();

方式二：使用Map直接映射 #

// 以用户ID为Key，RSocket连接为Value，线程安全的ConcurrentHashMap
private final ConcurrentHashMap<String, RSocketRequester> CLIENT_MAP = new ConcurrentHashMap<>();

2. 连接时绑定用户信息，并在断开时获取 #

用户连接时，通常会通过元数据（Metadata）或Payload携带身份信息，你可以在@ConnectMapping方法中解析这些信息，绑定到连接实例上，再在onClose()的回调中获取用户信息：

@ConnectMapping
public void onConnect(RSocketRequester requester, @Metadata String userId, @Payload String username) {
    log.info("RSocket连接建立：requester hashCode = {}", requester.hashCode());

    // 创建连接实体并绑定信息
    ClientConnection connection = new ClientConnection();
    connection.setUserId(userId);
    connection.setUsername(username);
    connection.setRequester(requester);

    requester.rsocket()
        .onClose()
        .doFirst(() -> CLIENT_CONNECTIONS.add(connection))
        .doOnError(error -> log.info("RSocket连接异常：{}", error.getMessage()))
        .doFinally(signalType -> {
            // 断开时直接获取用户信息
            log.info("用户 {}（ID：{}）已断开连接", connection.getUsername(), connection.getUserId());
            CLIENT_CONNECTIONS.remove(connection);
        })
        .subscribe();

    log.info("当前在线客户端数量：{}", CLIENT_CONNECTIONS.size());
}

说明：如果使用自定义元数据格式（如CompositeMetadata），需要配合对应的解码器解析指定MIME类型的元数据内容。

3. 覆盖所有断开场景 #

onClose()会触发所有连接终止场景：

• 用户主动关闭连接
• 网络波动导致连接中断
• 服务器主动断开连接
• 连接超时

在doFinally回调中，无论连接是正常关闭还是异常终止，都能安全获取用户信息并完成清理操作（如移除连接记录、更新在线状态等）。

4. 线程安全注意事项 #

原代码使用的ArrayList是非线程安全集合，多用户并发连接/断开时可能出现ConcurrentModificationException，建议替换为：

• CopyOnWriteArrayList（适合读多写少的场景）
• ConcurrentHashMap（适合按用户ID快速查找的场景）

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