Spring Boot STOMP集成AWS ActiveMQ高并发连接优化求助

针对负载测试中出现的大量WebSocket I/O Error，以及支持数千并发连接的需求，结合你的Spring Boot 2.7.0配置，给出以下优化点：

1. SSL配置优化 #

• 替换InsecureTrustManagerFactory.INSTANCE：生产环境使用不安全的信任管理器存在安全风险，且会因无证书缓存导致重复SSL握手，增加I/O开销。建议导入AWS ActiveMQ官方证书到信任库，或直接指定信任该证书：

  sslContext = SslContextBuilder.forClient()
          .sslProvider(SslProvider.JDK)
          .trustManager(new File("path/to/amazonmq-truststore.jks")) // 替换为实际证书路径
          .build();
  

• 启用SSL会话复用：通过配置握手超时和会话缓存，减少重复握手的资源消耗：

  .secure(ssl -> ssl.sslContext(sslContext)
          .handshakeTimeout(Duration.ofSeconds(10))
          .sessionTimeout(Duration.ofMinutes(5)))
  

2. Stomp Broker Relay连接池优化 #

当前TCP客户端未配置连接池，高并发下会频繁创建/销毁连接，引发I/O错误。为Reactor Netty配置连接池：

ConnectionProvider connectionProvider = ConnectionProvider.builder("amazonmq-stomp-pool")
        .maxConnections(200) // 根据ActiveMQ实例规格调整，参考AWS文档的连接数上限
        .pendingAcquireTimeout(Duration.ofSeconds(5))
        .build();

ReactorNettyTcpClient<byte[]> client = new ReactorNettyTcpClient<>(
        tcpClient -> tcpClient.remoteAddress(socketAddressSupplier())
                .connectionProvider(connectionProvider)
                .secure(ssl -> ssl.sslContext(sslContext)),
        new StompReactorNettyCodec());

3. 线程池配置优化 #

当前线程池存在队列无界、线程数配置不合理的问题，易引发内存溢出或线程耗尽：

public ThreadPoolTaskExecutor threadPoolTaskExecutor(String groupName) {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    int corePoolSize = Runtime.getRuntime().availableProcessors() * 2; // IO密集型任务建议设为2*CPU核心数
    executor.setCorePoolSize(corePoolSize);
    executor.setMaxPoolSize(corePoolSize + 30); // 避免线程数过度膨胀
    executor.setQueueCapacity(1000); // 设置有界队列，防止内存溢出
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 拒绝策略，避免任务丢失
    executor.setThreadGroupName(groupName);
    executor.setThreadNamePrefix(groupName);
    executor.initialize();
    return executor;
}

• 可根据业务场景区分入站和出站线程池参数：如果出站消息量更大，单独调整出站线程池的核心/最大线程数。

4. WebSocket容器调优 #

Spring Boot默认使用Tomcat作为容器，需调整WebSocket相关参数以支持高并发，在application.properties中添加：

# Tomcat WebSocket配置
server.tomcat.max-connections=10000 # 最大并发连接数，根据服务器资源调整
server.tomcat.websocket.max-text-message-buffer-size=65536 # 文本消息缓冲区大小
server.tomcat.websocket.max-binary-message-buffer-size=65536 # 二进制消息缓冲区大小
server.tomcat.accept-count=200 # 连接等待队列长度

5. Stomp Broker Relay心跳与重试配置 #

配置心跳机制避免空闲连接被AWS负载均衡或ActiveMQ断开，同时添加重试逻辑：

registry.enableStompBrokerRelay(stompBrokerConfigurationProperties.getRelay())
        // ... 原有配置
        .setSystemHeartbeat(new long[]{10000, 10000}) // 系统连接心跳：10秒发送一次
        .setClientHeartbeat(new long[]{10000, 10000}) // 客户端连接心跳：10秒发送一次
        .setRelayHost(stompBrokerConfigurationProperties.getHost()) // 显式指定主机
        .setRelayPort(stompBrokerConfigurationProperties.getPort())
        .setRecoveryInterval(5000); // 连接断开后重试间隔5秒

6. 错误处理增强 #

当前SSL异常仅打印日志，应终止启动避免不安全的服务运行：

catch (SSLException e) {
    log.error("Failed to initialize SSL context for STOMP broker relay", e);
    throw new IllegalStateException("STOMP broker relay SSL initialization failed", e);
}

7. AWS ActiveMQ侧配置建议 #

• 调整ActiveMQ实例规格：选择支持更高并发连接的实例类型（如mq.t3.large及以上），并根据需求调整最大连接数配置。
• 启用ActiveMQ连接池：在AmazonMQ控制台配置连接池参数，避免代理端连接耗尽。
• 监控ActiveMQ指标：通过CloudWatch监控Connections、MessageEnqueueRate等指标，定位瓶颈。

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