如何配置ActiveMQ Artemis客户端或服务端避免重复创建队列

问题描述 #

我部署了一个未预配置队列的嵌入式ActiveMQ Artemis 2.17.0 broker，期望客户端连接broker时，若队列不存在则自动创建。首次运行客户端时该逻辑正常，但再次连接broker时，客户端抛出错误：

errorType=QUEUE_EXISTS message=AMQ229019: Queue hornetq already exists on address router

请问是否可通过配置客户端或服务端，实现在队列已存在时不尝试重新创建队列？

我的代码 #

服务端代码

try {
    ActiveMQServer server = new ActiveMQServerImpl(new ConfigurationImpl()
            .setPersistenceEnabled(true)
            .setBindingsDirectory("./router/data/bindings")
            .setLargeMessagesDirectory("./router/data/large")
            .setPagingDirectory("./router/data/paging")
            .setJournalDirectory("./router/data/journal")
            .setSecurityEnabled(false)
            .addAcceptorConfiguration("tcp", "tcp://0.0.0.0:61617?protocols=CORE,AMQP"));
    server.start();
} catch (Exception ex) {
    System.err.println(ex);
}

客户端代码

ServerLocator serverLocator = ActiveMQClient.createServerLocator("tcp://127.0.0.1:61617");
ClientSessionFactory factory = serverLocator.createSessionFactory();
ClientSession session = factory.createSession();
session.createQueue(new QueueConfiguration("router::hornetq")
        .setAutoCreateAddress(Boolean.FALSE)
        .setAutoCreated(Boolean.FALSE)
        .setRoutingType(RoutingType.ANYCAST));
ClientProducer producer = session.createProducer("router::hornetq");
ClientMessage message = session.createMessage(true);
message.getBodyBuffer().writeString("Core Queue Message");
producer.send(message);
session.start();
ClientConsumer consumer = session.createConsumer("router::hornetq");
ClientMessage msgReceived = consumer.receive();
System.out.println("message = " + msgReceived.getBodyBuffer().readString());
session.close();

注：我使用全限定队列名（即router::hornetq）是因为router地址下存在多个队列。

解决方案 #

1. 客户端修改：添加「仅当不存在时创建」的逻辑 #

你当前的客户端代码每次运行都会无条件尝试创建队列，这就是第二次报错的原因。可以通过两种方式调整：

方式A：使用带ifNotExists参数的重载方法

直接改用ClientSession中支持ifNotExists参数的createQueue重载，当该参数设为true时，只有队列不存在才会执行创建操作，已存在则直接跳过：

// 替换原来的session.createQueue(...)代码
session.createQueue("router", "hornetq", RoutingType.ANYCAST, true);

这里的第四个参数true就是ifNotExists，完美匹配你的需求。

方式B：通过QueueConfiguration配置ifNotExists

如果你更习惯使用QueueConfiguration对象，只需添加一行setIfNotExists(true)即可：

session.createQueue(new QueueConfiguration("router::hornetq")
        .setAutoCreateAddress(Boolean.FALSE)
        .setAutoCreated(Boolean.FALSE)
        .setRoutingType(RoutingType.ANYCAST)
        .setIfNotExists(true)); // 新增这一行

配置后，创建请求会自动忽略已存在的队列，不会抛出QUEUE_EXISTS异常。

2. 服务端配置：开启自动创建，客户端无需手动创建 #

另一种更省心的方式是在服务端开启自动创建队列的功能，这样客户端完全不需要显式调用createQueue。当客户端发送消息或尝试消费队列时，服务端会自动检查队列是否存在，不存在则自动创建。

修改服务端的配置代码，添加自动创建相关参数：

try {
    ActiveMQServer server = new ActiveMQServerImpl(new ConfigurationImpl()
            .setPersistenceEnabled(true)
            .setBindingsDirectory("./router/data/bindings")
            .setLargeMessagesDirectory("./router/data/large")
            .setPagingDirectory("./router/data/paging")
            .setJournalDirectory("./router/data/journal")
            .setSecurityEnabled(false)
            // 新增自动创建配置
            .setAutoCreateQueues(true)
            .setAutoCreateAddresses(true)
            .addAcceptorConfiguration("tcp", "tcp://0.0.0.0:61617?protocols=CORE,AMQP"));
    server.start();
} catch (Exception ex) {
    System.err.println(ex);
}

配置完成后，客户端可以直接删除session.createQueue()这一行代码，直接进行消息的发送和消费操作即可。

补充说明 #

• 你的全限定队列名router::hornetq对应地址router和队列名hornetq，两种解决方案都能完美适配这种命名方式。
• 如果需要自定义队列的路由类型、持久化等属性，推荐使用客户端的ifNotExists方式；如果希望简化客户端逻辑，服务端自动创建是更优的选择。

内容的提问来源于stack exchange，提问作者kimathie