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

Question

如果我在 ActiveMQ Artemis test.A 和通配符队列 @9​​87654322@ 上创建队列，那么我可以向 test.A 发送消息，它也将被传递到 test.#。然而，令我惊讶的是，当我使用来自test.# 的消息时，该消息仍然存在于test.A 上

如何更改我的代码或配置以获得预期的行为？

示例代码：

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

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

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

• 按原样接受语义。
• 创建地址test.A没有队列，例如：

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

  这是一个完全有效的配置，但您将无法直接使用来自test.A 的消息，因为不存在这样的队列。您只能使用通配符地址上的队列中的消息。
• 在test.A 队列中将 purge-on-no-consumer 设置为 true，例如：

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

  此设置将允许队列在消费者连接时接收消息，但一旦最后一个消费者断开连接，队列将清除所有消息，并且只要没有消费者，消息就不会路由到它。

使用来自消费者的通配符来接收来自多个队列的消息是一个很好的便利功能，但它没有在 ActiveMQ Artemis 中实现。但是，创建多个消费者应该不是很困难。