多线程 JMS 客户端 ActiveMQ

Question

我正在使用以下代码为多个消费者创建多个 JMS 会话以使用消息。我的问题是代码以单线程方式运行。即使队列中存在消息，第二个线程也无法接收任何内容，并且只会继续轮询。第一个线程同时完成了第一批的处理并返回并使用剩余的消息。这里的用法有什么问题吗？

static {
    try {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://172.16.143.99:61616");
        connection = connectionFactory.createConnection();
        connection.start();
    } catch (JMSException e) {
        LOGGER.error("Unable to initialise JMS Queue.", e);
    }

}

public JMSClientReader(boolean isQueue, String name) throws QueueException {

    init(isQueue,name);
}

@Override
public void init(boolean isQueue, String name) throws QueueException
{

    // Create a Connection
    try {
        // Create a Session
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        if (isQueue) {
            destination = new ActiveMQQueue(name);// session.createQueue("queue");
        } else {
            destination = new ActiveMQTopic(name);// session.createTopic("topic");
        }
        consumer = session.createConsumer(destination);
    } catch (JMSException e) {
        LOGGER.error("Unable to initialise JMS Queue.", e);
        throw new QueueException(e);
    }
}

public String readQueue() throws QueueException {

    // connection.setExceptionListener(this);
    // Wait for a message
    String text = null;
    Message message;
    try {
        message = consumer.receive(1000);
        if(message==null)
            return "done";
        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            text = textMessage.getText();
            LOGGER.info("Received: " + text);
        } else {
            throw new JMSException("Invalid message found");
        }
    } catch (JMSException e) {
        LOGGER.error("Unable to read message from Queue", e);
        throw new QueueException(e);
    }


    LOGGER.info("Message read is " + text);
    return text;

}

Answer 1

您的问题是 prefetchPolicy。

persistent queues (default value: 1000)
non-persistent queues (default value: 1000)
persistent topics (default value: 100)
non-persistent topics (default value: Short.MAX_VALUE - 1)

所有消息都被分派给第一个连接的消费者，而当另一个消费者连接时，他不会收到消息，因此如果您有队列的并发消费者，则要更改此行为，您需要将 prefetchPolicy 设置为低于默认值的值。例如，将此jms.prefetchPolicy.queuePrefetch=1 添加到 activemq.xml 中的 uri 配置中，或者像这样在客户端 url 上设置它

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://172.16.143.99:61616?jms.prefetchPolicy.queuePrefetch=1");

建议使用较大的预取值以获得高性能和高 消息量。但是，对于较低的消息量，每个 消息需要很长时间来处理，预取应该设置为 1。 这确保了消费者一次只处理一条消息。 但是，将预取限制指定为零会导致消费者 一次轮询消息，而不是消息 推送给消费者。

看看http://activemq.apache.org/what-is-the-prefetch-limit-for.html

还有

http://activemq.apache.org/destination-options.html