不能同时从 ActiveMQ 嵌入式代理消费

Question

我正在寻找一种方法来将 10 条消息发布到 ActiveMQ 嵌入式代理，并在同一个 VM 上使用 JMS API.

同时使用它们

下面的代码存在某种竞争，因为有时它会并行处理 2、4、8 条消息，并挂起直到 latch.await 调用超时。

public final class ActiveMQJMSParallelTest {
    private static final Logger logger = LoggerFactory.getLogger(ActiveMQJMSParallelTest.class);
    private static final int numberOfMessages = 10;

    public static void main(final String[] args) throws Exception {
        final Properties props = new Properties();
        props.setProperty(Context.INITIAL_CONTEXT_FACTORY, "org.apache.activemq.jndi.ActiveMQInitialContextFactory");
        props.setProperty(Context.PROVIDER_URL, "vm://localhost?broker.persistent=false");
        props.setProperty("queue.parallelQueue", "parallelQueue");
        final Context jndiContext = new InitialContext(props);
        final ConnectionFactory connectionFactory = (ConnectionFactory) jndiContext.lookup("ConnectionFactory");
        final Destination destination = (Destination) jndiContext.lookup("parallelQueue");
        final Connection connection = connectionFactory.createConnection();
        Session session = null;
        try {
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            final MessageProducer producer = session.createProducer(destination);
            for (int i = 0; i < numberOfMessages; i++) {
                final TextMessage message = session.createTextMessage();
                message.setText("This is message " + (i + 1));
                producer.send(message);
                logger.info("Produced message: {}", message);
            }
            session.commit();
        } finally {
            if (session != null)
                session.close();
        }

        final CountDownLatch latch = new CountDownLatch(numberOfMessages);
        final ExecutorService pool = Executors.newFixedThreadPool(numberOfMessages);
        for (int i = 0; i < numberOfMessages; i++) {
            pool.submit(new Runnable() {
                @Override public void run() {
                    try {
                        final Connection connection = connectionFactory.createConnection();
                        connection.start();
                        final Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
                        final Queue destination = session.createQueue("parallelQueue");
                        final MessageConsumer consumer = session.createConsumer(destination);
                        final Message received = consumer.receive();
                        logger.info("Consuming message: {}", received);
                        latch.countDown();
                        latch.await(1, TimeUnit.MINUTES);
                        logger.info("Consumed message: {}", received);
                        session.close();
                        connection.close();
                    } catch(Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        latch.await(10, TimeUnit.MINUTES);
        jndiContext.close();
    }
}

有人可以为此任务想出工作代码吗？

Answer 1

如果你想确保每个消费者都有机会一次获取一条消息，那么你应该使用零预取值，这样代理就不会尝试为第一个消费者发送达到预取限制等他们到达时。

查看 documentation 页面上的预取是如何工作的。

不能同时从 ActiveMQ 嵌入式代理消费

Cannot concurrently consume from ActiveMQ embedded broker

java

activemq

jms