Spring WebSocket(STOMP)对接RabbitMQ外部代理的连接池实现问询

一、Spring框架原生支持情况 #

Spring官方的StompBrokerRelayMessageHandler（即你配置中enableStompBrokerRelay对应的处理器）默认没有提供连接池机制，它的设计逻辑是为每个客户端的STOMP会话创建独立的RabbitMQ STOMP连接，这也是当前每个客户端对应一个RabbitMQ连接的原因。

二、可行实现方案 #

1. 自定义连接池化的StompBrokerRelay #

核心思路是替换默认处理器的底层连接管理逻辑，自行实现连接池复用RabbitMQ STOMP连接。

关键步骤：

• 实现STOMP连接池：可基于Apache Commons Pool2或自定义简单对象池，管理StompConnection实例（Spring内部StompClientSupport相关实现）。
• 重写StompBrokerRelayMessageHandler的createSession方法：创建客户端会话时从连接池获取空闲连接，会话关闭时将连接归还池内。
• 配置自定义处理器：放弃enableStompBrokerRelay，手动注册自己的连接池化代理处理器。

示例代码片段：

// 自定义连接池实现（简化版）
public class StompConnectionPool {
    private final GenericObjectPool<StompConnection> pool;

    public StompConnectionPool(StompConnectionFactory factory) {
        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxTotal(10); // 最大连接数
        this.pool = new GenericObjectPool<>(factory, poolConfig);
    }

    public StompConnection borrowConnection() throws Exception {
        return pool.borrowObject();
    }

    public void returnConnection(StompConnection connection) {
        pool.returnObject(connection);
    }
}

// 重写会话创建逻辑的自定义处理器
public class PooledStompBrokerRelayMessageHandler extends StompBrokerRelayMessageHandler {
    private final StompConnectionPool connectionPool;

    public PooledStompBrokerRelayMessageHandler(MessageBrokerRegistry registry, StompConnectionPool pool) {
        super(registry.getApplicationDestinationPrefixes(), registry.getUserDestinationPrefixes());
        this.connectionPool = pool;
        // 复制原有代理配置
        setRelayHost(registry.getRelayHost());
        setRelayPort(registry.getRelayPort());
        setClientLogin(registry.getClientLogin());
        setClientPasscode(registry.getClientPasscode());
        // 其他配置按需复制
    }

    @Override
    protected StompSession createSession(StompHeaderAccessor connectHeaders) throws Exception {
        StompConnection connection = connectionPool.borrowConnection();
        StompSession session = connection.connect(connectHeaders);
        // 会话关闭时归还连接
        session.setSessionClosedCallback(() -> connectionPool.returnConnection(connection));
        return session;
    }
}

配置类中替换默认处理器：

@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
    registry.setPreservePublishOrder(true)
            .setApplicationDestinationPrefixes(APP_PREFIXES);
    
    // 初始化连接池
    StompConnectionFactory connectionFactory = new StompConnectionFactory(rabbitMQHost, rabbitMQPort, rabbitMQUsername, rabbitMQPassword);
    StompConnectionPool connectionPool = new StompConnectionPool(connectionFactory);
    
    // 注册自定义池化处理器
    registry.setMessageHandler(new PooledStompBrokerRelayMessageHandler(registry, connectionPool));
    
    // 原有通道配置保留
    registry.configureBrokerChannel()
            .interceptors(new LoggingInterceptor(), new EscapeSlashesInterceptor())
            .taskExecutor().corePoolSize(1).maxPoolSize(MAX_WORKERS_COUNT).queueCapacity(TASK_QUEUE_SIZE);
}

2. 改用RabbitMQ AMQP代理模式（间接复用连接） #

放弃直接的STOMP中继，改为Spring WebSocket接收消息后，通过AMQP连接池将消息发送到RabbitMQ，再由RabbitMQ STOMP插件转发给客户端，利用成熟的AMQP连接池实现复用。

关键步骤：

• 关闭enableStompBrokerRelay，启用enableSimpleBroker作为内部临时代理。
• 配置Spring AMQP的CachingConnectionFactory（默认支持连接池）和RabbitTemplate。
• 编写消息处理器，将WebSocket消息通过RabbitTemplate发送到RabbitMQ的STOMP交换器（默认/exchange/amq.topic）。

示例配置：

// AMQP连接池配置
@Bean
public CachingConnectionFactory rabbitConnectionFactory() {
    CachingConnectionFactory factory = new CachingConnectionFactory(rabbitMQHost, rabbitMQPort);
    factory.setUsername(rabbitMQUsername);
    factory.setPassword(rabbitMQPassword);
    factory.setVirtualHost(rabbitMQVirtualHost);
    // 连接池参数配置
    factory.setConnectionCacheSize(10); // 最大复用连接数
    factory.setChannelCacheSize(50); // 最大通道数
    return factory;
}

@Bean
public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
    return new RabbitTemplate(connectionFactory);
}

// WebSocket配置
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
    registry.setPreservePublishOrder(true)
            .setApplicationDestinationPrefixes(APP_PREFIXES)
            .enableSimpleBroker("/topic"); // 内部临时代理
    
    registry.configureBrokerChannel()
            .interceptors(new LoggingInterceptor(), new EscapeSlashesInterceptor())
            .taskExecutor().corePoolSize(1).maxPoolSize(MAX_WORKERS_COUNT).queueCapacity(TASK_QUEUE_SIZE);
}

// 消息转发控制器
@Controller
public class MessageRelayController {
    private final RabbitTemplate rabbitTemplate;

    public MessageRelayController(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @MessageMapping("/send/{topic}")
    public void relayMessage(@DestinationVariable String topic, String message) {
        // 发送到RabbitMQ STOMP主题交换器
        rabbitTemplate.convertAndSend("/exchange/amq.topic", topic, message);
    }
}

这种方式下，Spring应用与RabbitMQ的连接由CachingConnectionFactory管理复用，不会随客户端数量线性增长。

三、注意事项 #

• 自定义STOMP连接池时，需处理连接心跳、异常回收逻辑，避免无效连接占用池资源。
• 使用AMQP代理模式时，需确保RabbitMQ已启用STOMP插件，并正确配置交换器与队列的绑定。
• 两种方案各有优劣：自定义STOMP连接池更贴近原有中继逻辑，AMQP代理模式利用成熟组件，维护成本更低。

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