java - 使用 camel 和 activemq 到 "pause"路由的正确方法是什么？

Question

我的基本问题是，在任何 1 小时的时间段内，我只能处理来 self 的一个队列(跨所有机器)的 7000 条消息。我看不到用 camel 或 activemq 来做到这一点的方法，所以我求助于实现我自己的路由停止/启动逻辑。我看到了很多方法可以做到这一点，我已经尝试了其中的一些(只是遇到了问题)。

1. camelContext.stopRoute(route):这在停止处理消息时有效，但是当我调用 camelContext.startRoute(route) 时，它泄漏了一个 tcp 连接最终导致 activemq 服务器达到其极限并死亡。
2. camelContext.suspendRoute(route):这也会停止处理消息并且不会泄漏连接，但它似乎会杀死在以下情况下不会重新激活的活跃消费者(在管理面板中可见)我调用 camelContext.resumeRoute(route)。我认为这最终可能导致该队列中根本没有消息被处理，即使我恢复也是如此。
3. 实现自定义 RoutePolicy。公平地说，我还没有尝试过这个，但根据我在上面选择的暂停方法，它似乎会成为我遇到的同样问题的牺牲品。

有没有解决这个问题的方法我还没有遇到过？

Answer 1

与其停止路线，我建议使用 Throttler EIP .

from("jms:queue:inbox")
    .throttle(7000)
    .timePeriodMillis(1000*60*60)
    .to("log:result", "mock:result");

上面的例子会限制在 jms:queue:inbox 上收到的消息，然后再发送到 mock:result 确保在任何 1 小时内最多发送 7000 条消息窗口。

或者，为了更细粒度的控制，您可以定义限制路由策略，如 Camel 的 throttling example 所示。 :

<route routePolicyRef="myPolicy">
    <from uri="jms:queue:inbox"/>
    <transacted/>
    <to uri="log:+++JMS +++?groupSize=100"/>
    <to ref="foo"/>
</route>

节流警察定义如下:

<bean id="myPolicy" class="org.apache.camel.impl.ThrottlingInflightRoutePolicy">
    <property name="scope" value="Context"/>
    <!-- when we hit > 20 inflight exchanges then kick in and suspend the routes -->
    <property name="maxInflightExchanges" value="20"/>
    <!-- when we hit lower than 10% of the max = 2 then kick in and resume the routes the default percentage is 70% but in this demo we want a low value -->
    <property name="resumePercentOfMax" value="10"/>
    <!-- output throttling activity at WARN level -->
    <property name="loggingLevel" value="WARN"/>
</bean>

编辑 1:

如果需要全局限流，那么可以先让一个消费者读取消息，按照上面的方法对所有消息进行限流，然后重新发送到另一个队列，让其重新读取和处理他们由 >= 1 分布式消费者。

编辑 2:

或者，您可以实现自己的ThrottlingInflightRoutePolicy 来访问保存处理信息的中央数据库。这样，您就不需要“单节点主节流器”。但是，数据库也可能是单点故障。