Camel Throttling使用correlationExpression按target_queue分组限流咨询

问题根因 #

你遇到的限流未按分组生效的问题，主要有三个常见原因：

• 你使用的Apache Camel版本低于2.16：基于correlationExpression的分组限流功能是Camel 2.16版本才正式上线的特性，低版本会忽略该配置，直接走全局限流逻辑。
• XML配置的节点顺序错误：XML DSL中要求correlationExpression必须作为throttle节点的第一个子元素，放在限流阈值配置之前，否则配置不会被限流处理器识别。
• 你当前配置的消费队列URI存在多余的>符号，会导致路由无法正常消费消息，需要修正。

修正后的路由配置 #

<camel:route
    id="com.xxx.camel.route.throttle"
    autoStartup="true">
    <!-- 修正URI中的多余>符号 -->
    <camel:from
        uri="activemq:queue:com.xxx.camel.route.throttle?destination.consumer.prefetchSize=0" />
    <camel:throttle
        timePeriodMillis="10000"
        asyncDelayed="true">
        <!-- 调整顺序：correlationExpression放在最前面 -->
        <camel:correlationExpression>
            <camel:header>target_queue</camel:header>
        </camel:correlationExpression>
        <!-- 限流阈值配置放在correlationExpression之后 -->
        <camel:header>messages_per_10s_period</camel:header>
        <camel:routingSlip>
            <camel:header>target_queue</camel:header>
        </camel:routingSlip>
    </camel:throttle>
</camel:route>

配置验证建议 #

• 先确认消息流转过程中target_queue header没有被修改或丢失，可在路由入口加日志打印该header值验证。
• 如需固定单分组限流阈值，可将<camel:header>messages_per_10s_period</camel:header>替换为固定常量，比如<camel:constant>50</camel:constant>，代表每个target_queue分组每10秒最多放行50条消息。
• 开启Camel DEBUG日志，观察org.apache.camel.processor.Throttler类的输出，可直接看到每个分组的限流计数，确认分组逻辑是否符合预期。

内容的提问来源于stack exchange，提问作者Jonathan S. Fisher