如何提升Spring Boot JMS消息消费速率？

一、开启并发消费（核心吞吐量提升手段） #

默认DefaultMessageListenerContainer的并发消费者数量为1，单线程处理是当前速率极低的关键原因之一。通过以下配置开启并发消费：

修改JMS容器配置，设置初始并发数与最大并发数：

@Bean
public DefaultMessageListenerContainer listenerContainer(MessageListenerAdapter messageListener,
        @Qualifier("sourceConnection") ConnectionFactory listenerConnectionFactory) {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(listenerConnectionFactory);
    container.setDestinationName(jmsSourceQueue);
    container.setMessageListener(messageListener);
    container.setSessionTransacted(true);
    container.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
    container.setRecoveryInterval(30000);
    // 初始并发消费者数，根据服务器配置调整（如5）
    container.setConcurrentConsumers(5);
    // 最大并发消费者数，峰值时自动扩容（如10）
    container.setMaxConcurrentConsumers(10);
    // 预取数设置，减少客户端与JMS服务器交互次数（根据消息大小调整，如100）
    container.setPrefetchSize(100);
    return container;
}

注意事项：

• 确保MyMessageService、FileIOHelper、XmlUtil均为线程安全：若FileIOHelper按accountNumber写入不同文件，无需额外锁；若写入同一文件，需添加同步锁避免内容混乱。
• 并发数不宜过大，避免耗尽服务器CPU、内存或磁盘IO，建议按服务器CPU核数的2倍设置。

二、优化文件写入操作（解决IO瓶颈） #

同步文件写入属于IO密集型操作，会阻塞消息处理线程，可从以下方向优化：

1. 异步执行文件写入 #

将文件写入从消息监听线程剥离，交给异步线程池处理，让核心逻辑快速执行：

首先配置异步线程池：

@Bean
public TaskExecutor fileWriteExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(8);
    executor.setMaxPoolSize(16);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("file-writer-");
    executor.initialize();
    return executor;
}

修改MyMessageListener，异步执行文件写入：

@Autowired
private TaskExecutor fileWriteExecutor;

@Override
public void onMessage(Message message) {
    String messageContent = null;
    try {
        // 消息解析逻辑（省略）...

        if (messageContent != null) {
            String finalMessageContent = messageContent;
            // 异步写入入站文件
            fileWriteExecutor.execute(() -> FileIOHelper.writeInboundXmlToFile(finalMessageContent));
            
            String accountNumber = XmlUtil.extractAccountNumber(messageContent);
            final String xmlMessageTransformed = messageService.transformXmlMessageToOldSchema(messageContent);
            
            if (!xmlMessageTransformed.isEmpty()) {
                String finalXml = xmlMessageTransformed;
                String finalAccount = accountNumber;
                // 异步写入转换后文件
                fileWriteExecutor.execute(() -> FileIOHelper.writeTransformedXmlToFile(finalAccount, finalXml));
                
                Map<String, String> outboundHeaderProperties = messageService.createJMSHeaderProperties(message);
                // 若出站发布耗时，也可异步执行（需保证消息可靠性）
                messageService.publishMessageToOutboundTopic(xmlMessageTransformed, outboundHeaderProperties);
            } else {
                String finalUnprocessed = messageContent;
                // 异步写入未处理文件
                fileWriteExecutor.execute(() -> FileIOHelper.writeUnprocessedXmlToFile(finalUnprocessed));
                log.error(String.format("Failed transformation of message account# %s", accountNumber));
            }
            message.acknowledge();
        }
    } catch (Exception e) {
        log.error("Message processing failed", e); // 打印完整堆栈便于排查
    }
}

注意：异步写入需添加失败处理逻辑（如记录到失败日志表），避免丢失关键数据。

2. 优化文件IO底层实现 #

• 检查FileIOHelper：若每次写入都打开/关闭文件流，改为复用缓冲流（如BufferedWriter），减少IO操作次数；若写入不同文件，优化目录创建逻辑，避免重复创建。
• 使用NIO异步API：如Files.writeAsync()替代同步写入，提升IO效率。

三、优化消息处理其他环节 #

• XML解析与转换优化：
  • 若transformXmlMessageToOldSchema使用XSLT转换，缓存XSLT模板实例，避免每次重新加载。
  • 改用高效XML解析库（如Jackson XML、StAX）替代DOM解析，减少内存占用与解析耗时。
• 出站消息发布优化：若publishMessageToOutboundTopic为同步操作，可放入异步线程池执行，减少消息处理线程阻塞时间。

四、调整JMS事务与确认机制 #

当前配置setSessionTransacted(true)，每次消息处理都会触发事务提交，带来额外开销：

• 若业务允许关闭事务，仅保留CLIENT_ACKNOWLEDGE模式，在核心逻辑完成后手动acknowledge。
• 若必须保留事务，将异步操作（如文件写入）移出事务边界，仅在事务中执行核心解析、转换与发布逻辑，缩短事务持有时间。

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