Apache Camel 3.14.9：CLIENT_ACKNOWLEDGE模式下批量消息异常处理咨询

关键实现思路 #

针对你的核心诉求，核心实现要点如下：

• CLIENT_ACKNOWLEDGE模式：配置JMS组件启用该模式，关闭自动确认，仅在整个批次处理成功后手动确认消息，确保路由崩溃时输入队列保留未处理的聚合批次。
• 批次异常终止：通过Camel的异常拦截机制，捕获处理器抛出的异常后立即终止当前批次的后续处理，同时将异常批次转发至指定错误队列。
• 消息可靠性保障：基于聚合批次维度的确认逻辑，替代单条消息的自动确认，避免部分处理失败导致的消息丢失。

1. 「全部消息」与「仅聚合批次」的区别 #

• 全部消息：指输入队列中所有独立的原始消息。若使用自动确认模式，每条消息被Camel接收后就会从队列移除——哪怕后续聚合或处理失败，消息也无法找回。
• 仅聚合批次：指经过Aggregator EIP聚合后的完整消息组。启用CLIENT_ACKNOWLEDGE模式时，我们只在整个批次处理完成且无异常时，才确认这个聚合批次的消息；如果中途出现异常或路由崩溃，整个聚合批次会保留在输入队列中，不会丢失。

简单来说，前者是单条消息维度的确认，后者是聚合后的批次维度确认，后者更适合批量处理场景的消息可靠性保障。

2. 完整实现代码 #

依赖配置（pom.xml） #

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.camel.springboot</groupId>
        <artifactId>camel-spring-boot-starter</artifactId>
        <version>3.20.2</version> <!-- 请适配你的Spring Boot版本 -->
    </dependency>
    <dependency>
        <groupId>org.apache.camel.springboot</groupId>
        <artifactId>camel-jms-starter</artifactId>
        <version>3.20.2</version>
    </dependency>
</dependencies>

JMS组件配置（启用CLIENT_ACKNOWLEDGE） #

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.component.jms.JmsComponent;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class JmsConfig {

    @Bean
    public JmsComponent jmsComponent() {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setBrokerURL("tcp://localhost:61616");
        connectionFactory.setUserName("admin");
        connectionFactory.setPassword("admin");
        
        JmsComponent jmsComponent = new JmsComponent();
        jmsComponent.setConnectionFactory(connectionFactory);
        // 启用CLIENT_ACKNOWLEDGE模式
        jmsComponent.setAcknowledgementModeName("CLIENT_ACKNOWLEDGE");
        // 禁用自动确认，由业务逻辑手动控制消息确认时机
        jmsComponent.setAutoAcknowledge(false);
        return jmsComponent;
    }
}

核心路由实现 #

import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.aggregate.GroupedBodyAggregationStrategy;
import org.springframework.stereotype.Component;

@Component
public class BatchProcessingRoute extends RouteBuilder {

    @Override
    public void configure() throws Exception {
        // 局部异常处理：拦截批次处理中的所有异常
        onException(Exception.class)
                .log(LoggingLevel.ERROR, "批次处理触发异常，终止后续处理，转发至错误队列: ${exception.message}")
                .handled(true)
                // 将异常批次转发至错误队列
                .to("jms:queue:error.batch.queue")
                // 终止当前批次的后续处理流程
                .stop();

        from("jms:queue:input.batch.queue")
                // 聚合逻辑：按批次大小（5条）或超时（5秒）聚合消息
                .aggregate(constant(true), new GroupedBodyAggregationStrategy())
                .completionSize(5)
                .completionTimeout(5000)
                .log(LoggingLevel.INFO, "聚合完成，当前批次包含 ${body.size()} 条消息")
                
                // 拆分聚合批次为单条消息
                .split(body())
                .log(LoggingLevel.INFO, "拆分出单条消息: ${body}")
                
                // 重排逻辑：按messageId字段排序，启用批次重排
                .resequence(header("messageId"))
                .batch(true)
                .size(5)
                .log(LoggingLevel.INFO, "重排后发送至处理器: ${body}")
                
                // 自定义消息处理器
                .process(exchange -> {
                    String message = exchange.getIn().getBody(String.class);
                    // 模拟异常场景：消息包含"error"时抛出异常
                    if (message.contains("error")) {
                        throw new RuntimeException("处理失败：" + message);
                    }
                    // 正常处理逻辑
                    exchange.getIn().setBody("已处理: " + message);
                })
                
                // 批次处理成功后，手动确认输入队列的聚合批次消息
                .process(exchange -> {
                    javax.jms.Session session = exchange.getIn().getHeader(Exchange.JMS_SESSION, javax.jms.Session.class);
                    if (session != null) {
                        session.acknowledge();
                    }
                })
                .log(LoggingLevel.INFO, "批次处理完成，已确认消息")
                
                // 转发至输出队列
                .to("jms:queue:output.batch.queue");
    }
}

关键细节说明 #

1. CLIENT_ACKNOWLEDGE生效逻辑：通过JmsComponent关闭自动确认后，只有在批次所有消息处理成功时，才会调用session.acknowledge()确认聚合批次消息。如果路由崩溃或处理异常，消息不会被确认，会一直保留在输入队列中。
2. 批次异常终止：onException()拦截异常后，调用stop()立即终止当前批次的后续处理，避免剩余消息被继续执行；同时将异常批次转发至错误队列，方便后续排查。
3. 重排器批次模式：resequence()启用batch(true)后，会等待整个聚合批次的消息都到达后再进行排序，确保批次内消息按指定规则有序处理，配合异常终止逻辑实现“任一消息异常则停止批次处理”的诉求。

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