多线程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配置中，或像这样在客户端网址上进行设置

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