java - ActiveMQ 节流消费者

Question

我想在 hornetq 中对 activeMQ 中某个队列的消费者进行节流（对于 jboss，这是使用 mdb 消费者定义上的注释来实现的）。我在activemq的文档中找不到任何类似的东西，我找到的最接近的是这个

consumer.recvDelay   0 ms    Pause consumer for recvDelay milliseconds with each message (allows consumer throttling).

来自：http ://activemq.apache.org/activemq-performance-module-users-manual.html

但是我找不到如何在java中做到这一点。

提前致谢，

问候。

编辑：这是 ActiveMQManager 代码和使用者代码：

public class ActiveMQManager {

    private static ActiveMQConnectionFactory CONNECTION_FACTORY;

    public static Connection CONNECTION;

    public static Session SESSION;

    public static Destination TEST_QUEUE;

    public static void start() {
        try {

            CONNECTION_FACTORY = new ActiveMQConnectionFactory("vm://localhost");

            CONNECTION = CONNECTION_FACTORY.createConnection();
            CONNECTION.start();

            SESSION = CONNECTION.createSession(false,
                    Session.CLIENT_ACKNOWLEDGE);

            TestClient testClient = new TestClient();

            TEST_QUEUE = SESSION.createQueue("TEST.QUEUE");

            MessageConsumer testConsumer = SESSION.createConsumer(TEST_QUEUE);
            test.setMessageListener(testClient);

        } catch (Exception e) {
        }
    }

    public static void stop() {
        try {
            // Clean up
            SESSION.close();
            CONNECTION.close();
        } catch (JMSException e) {
            log.error(e);
        }
    }
}

消费者代码非常简单（对于这个例子）：

public class TestConsumer implements MessageListener {

    @Override
    public void onMessage(Message message) {
        //Do something with the message
    }

}

score 7 · Accepted Answer

这取决于所使用的消费者技术......但这里有一些选择

您可以手动在您的消费者代码中引入延迟（不是一门精确的科学，但这会限制吞吐量）
您还可以通过设置 JMS 连接的 maxConcurrentConsumers 属性来控制使用者使用的线程数……也就是说，这不会限制消息吞吐量，只是限制使用者使用的并发级别
更好的是，您可以使用限制器 EIP 实现设置每个时间段要消耗的确切消息数

例如，使用Camel Throttler很简单

from("activemq:queueA").throttle(10).to("activemq:queueB")