Camel结合ThrottlingExceptionRoutePolicy避免消息入DLQ的方案咨询

你的问题核心在于ThrottlingExceptionRoutePolicy是在Camel默认错误处理流程完成后才会触发路由暂停，这就导致触发故障的消息已经被移到DLQ了，而你希望这个消息能保留在原队列（或者不进入DLQ）同时暂停路由。下面是几个可行的配置方案和自定义实现思路：

方案1：使用onException结合手动触发路由暂停 #

如果不想自定义整个错误处理器，这是最直接的配置化方案——用Camel的onException DSL针对Web服务宕机的特定异常做拦截，先触发路由暂停再回滚事务，避免消息进入DLQ：

<camel:route id="someroute" routePolicyRef="throttlingExceptionRoutePolicy" errorHandlerRef="myTransactionErrorHandlerErrorHandler">
    <camel:from uri="activemq:inputQueue" />
    <camel:transacted />
    
    <!-- 针对Web服务不可达的异常配置专属处理逻辑 -->
    <camel:onException>
        <!-- 捕获HTTP服务异常，根据实际抛出的异常类调整 -->
        <camel:exception>org.apache.camel.component.http.HttpOperationFailedException</camel:exception>
        <!-- 匹配服务宕机相关的状态码：503服务不可用、504网关超时 -->
        <camel:when>
            <camel:simple>${exception.statusCode} == 503 || ${exception.statusCode} == 504</camel:simple>
        </camel:when>
        <!-- 手动调用路由策略的异常处理方法，立即暂停路由 -->
        <camel:bean ref="throttlingExceptionRoutePolicy" method="onExceptionOccurred(${route}, ${exchange}, ${exception})"/>
        <!-- 回滚事务，让消息留在原队列而不是进入DLQ -->
        <camel:rollback/>
        <!-- 标记异常已处理，终止后续默认错误流程 -->
        <camel:handled>
            <camel:constant>true</camel:constant>
        </camel:handled>
    </camel:onException>

    <camel:bean ref="afterQueueProcessor" />
    <camel:setHeader headerName="CamelHttpMethod">
        <camel:constant>POST</camel:constant>
    </camel:setHeader>
    <camel:setHeader headerName="Content-Type">
        <camel:constant>application/xyz</camel:constant>
    </camel:setHeader>
    <camel:to uri="http://localhost:8080/some-ws/newOrder?orderId=dd&amp;productName=bb&amp;quantity=1" />
</camel:route>

这个方案的关键逻辑：

• 精准拦截Web服务宕机的异常，避免误触发其他异常的处理
• 手动触发路由暂停，绕过原Policy的"错误处理后再触发"逻辑
• 回滚事务确保消息留在原队列，同时终止默认错误流程，阻止消息进入DLQ

方案2：自定义事务错误处理器（更灵活） #

如果需要更复杂的异常判断逻辑，可以自定义一个继承TransactionErrorHandler的处理器，在异常处理的最前端干预流程：

1. 实现自定义错误处理器 #

public class CustomTransactionErrorHandler extends TransactionErrorHandler {

    private final ThrottlingExceptionRoutePolicy throttlingPolicy;

    public CustomTransactionErrorHandler(CamelContext camelContext, Processor processor,
                                         TransactionErrorHandlerBuilder builder,
                                         ThrottlingExceptionRoutePolicy throttlingPolicy) {
        super(camelContext, processor, builder);
        this.throttlingPolicy = throttlingPolicy;
    }

    @Override
    public void handleException(Exchange exchange, Throwable exception) throws Exception {
        // 检测是否为Web服务宕机相关异常（根据实际情况调整判断逻辑）
        boolean isServiceDown = false;
        if (exception instanceof HttpOperationFailedException) {
            HttpOperationFailedException httpEx = (HttpOperationFailedException) exception;
            isServiceDown = httpEx.getStatusCode() == 503 || httpEx.getStatusCode() == 504;
        } else if (exception instanceof ConnectException) {
            // 处理连接超时的情况
            isServiceDown = true;
        }

        if (isServiceDown) {
            // 立即触发路由暂停
            throttlingPolicy.onExceptionOccurred(exchange.getContext().getRoute("someroute"), exchange, exception);
            // 标记事务回滚，消息留在原队列
            exchange.setRollbackOnly();
            return;
        }

        // 其他异常走默认错误处理流程
        super.handleException(exchange, exception);
    }
}

2. 配置Spring Bean并引用到路由 #

<!-- 注册自定义错误处理器 -->
<bean id="customTransactionErrorHandler" class="com.yourpackage.CustomTransactionErrorHandler">
    <constructor-arg ref="camelContext"/>
    <constructor-arg ref="afterQueueProcessor"/>
    <constructor-arg>
        <bean class="org.apache.camel.builder.errorhandler.TransactionErrorHandlerBuilder">
            <property name="transactionManager" ref="jmsTransactionManager"/>
            <!-- 保留你原有错误处理器的配置 -->
        </bean>
    </constructor-arg>
    <constructor-arg ref="throttlingExceptionRoutePolicy"/>
</bean>

<!-- 修改路由引用自定义错误处理器 -->
<camel:route id="someroute" routePolicyRef="throttlingExceptionRoutePolicy" errorHandlerRef="customTransactionErrorHandler">
    <!-- 原有路由内容不变 -->
</camel:route>

关键注意事项 #

• 确保你的ActiveMQ队列启用了事务支持，否则回滚操作无法让消息留在原队列
• 异常判断逻辑要和实际场景匹配：比如Web服务宕机可能抛出ConnectException或SocketTimeoutException，需要根据实际捕获的异常类调整
• 测试时模拟Web服务宕机场景，验证路由是否暂停、消息是否留在原队列，避免出现预期外的行为

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