在ActiveMQ Artemis中使用通配符队列时未使用的消息

Question

import org.apache.activemq.artemis.api.core.*;
import org.apache.activemq.artemis.api.core.client.*;
import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
import org.apache.activemq.artemis.core.server.embedded.EmbeddedActiveMQ;

public class Main {
    static String AMQ = "vm://0";

    public static void main(String[] args) throws Exception {
        EmbeddedActiveMQ server = null;
        try {
            server = createEmbeddedBroker();

            var serverLocator = ActiveMQClient.createServerLocator(AMQ);
            var clientSessionFactory = serverLocator.createSessionFactory();

            createQueues(clientSessionFactory);

            // queues are empty on creation
            try (var session = clientSessionFactory.createSession()) {
                assertQueueLength(session, "test.#", 0);
                assertQueueLength(session, "test.A", 0);
            }
            
            sendMessage(clientSessionFactory, "test.A");

            // expect message is delivered to both
            try (var session = clientSessionFactory.createSession()) {
                assertQueueLength(session, "test.#", 1);
                assertQueueLength(session, "test.A", 1);
            }

            consumeMessage(clientSessionFactory, "test.#");

            // expect message is consumed from both
            try (var session = clientSessionFactory.createSession()) {
                assertQueueLength(session, "test.#", 0); // ok - message gone
                assertQueueLength(session, "test.A", 0); // fails! 
            }

        } finally {
            if (server != null) server.stop();
        }
    }

    private static EmbeddedActiveMQ createEmbeddedBroker() throws Exception {
        var config = new ConfigurationImpl();
        config.addAcceptorConfiguration("vm", AMQ);
        config.setSecurityEnabled(false);
        config.setPersistenceEnabled(false);

        var server = new EmbeddedActiveMQ();
        server.setConfiguration(config);
        server.start();

        return server;
    }

    private static void createQueues(ClientSessionFactory csf)  throws Exception {
        var session = csf.createSession();

        /*
        <address name="test.A">
            <anycast>
                <queue name="test.A" />
            </anycast>
        </address>
        */
        var testA = new QueueConfiguration("test.A")
            .setRoutingType(RoutingType.ANYCAST)
            .setAddress("test.A");
        session.createQueue(testA);

        /*
        <address name="test.#">
            <anycast>
                <queue name="test.#" />
            </anycast>
        </address>
        */
        var testWildcard = new QueueConfiguration("test.#")
            .setRoutingType(RoutingType.ANYCAST)
            .setAddress("test.#");
        session.createQueue(testWildcard);

        // also tried to create address without a queue, but the message to test.A is not delivered to test.#
        // session.createAddress(new SimpleString("test.#"), RoutingType.ANYCAST, false);
    }

    private static void sendMessage(ClientSessionFactory csf, String queue) throws Exception {
        var session = csf.createSession();
        var producer = session.createProducer(queue);
        producer.send(session.createMessage(true));
        producer.close();
        session.close();
    }

    private static void consumeMessage(ClientSessionFactory csf, String queue) throws Exception {
        var session = csf.createSession();
        var consumer = session.createConsumer(queue);
        consumer.setMessageHandler(message -> {
            try {
                log("Consuming one message from " + queue);
                message.acknowledge();
                log("Consumed one message from " + queue);
            } catch (ActiveMQException e) {
                throw new IllegalStateException(e);
            }
        });
        session.start();

        Thread.sleep(1000); // hack to wait

        consumer.close();
        session.close();
    }

    private static void assertQueueLength(ClientSession session, String queue, long expected) throws Exception {
        long actual = session.queueQuery(SimpleString.toSimpleString(queue)).getMessageCount();
        if (actual != expected) {
            throw new IllegalStateException("Queue " + queue + " has " + actual + " messages. Expected " + expected);
        } else {
            log("Queue " + queue + " has " + actual + " messages as expected");
        }
    }

    private static void log(String msg) {
        System.out.println(">>> " + msg);
    }
}

依赖关系：

org.apache.activemq:artemis-core-client:2.17.0
org.apache.activemq:artemis-server:2.17.0

Answer 1

你所看到的是预期的行为。这里要记住的关键是，您使用的是通配符路由，而不是通配符消耗。使用通配符路由，消息不仅被路由到显式发送消息的地址的队列，而且还被路由到匹配通配符地址的任何队列。消息路由到的每个队列都有自己的消息副本。

通配符路由是在考虑多播（即pub/sub）用例（如分层主题）的情况下实现的，但如果您想将其与选播一起使用，有以下几个选项：

• Accept the semantics as-is.
• Create the address test.A without a queue, e.g.:

  session.createAddress(SimpleString.toSimpleString("test.A"), RoutingType.ANYCAST, false);
  

  var testA = new QueueConfiguration("test.A")
      .setRoutingType(RoutingType.ANYCAST)
      .setPurgeOnNoConsumers(true)
      .setAddress("test.A");
  

  使用使用者的通配符从多个队列接收消息是一个非常方便的特性，但在ActiveMQ Artemis中没有实现。然而，创建多个消费者应该不是很难。