ActiveMQ Classic结合AMQP 1.0使用带通配符Routing Key问题

核心原理说明 #

ActiveMQ Classic对AMQP 1.0的主题地址支持通配符映射，直接沿用ActiveMQ原生的主题路由规则即可：

• * 匹配单个层级（如TEST.*可匹配TEST.ORDER、TEST.PAYMENT）
• > 匹配所有后续层级（如TEST.>可匹配TEST.ORDER.CREATE、TEST.PAYMENT.SUCCESS）
  Qpid JMS客户端会将AMQP主题地址直接传递给ActiveMQ，由后者处理通配符路由逻辑，实现类似VirtualTopic的多消费者订阅效果。

具体配置步骤 #

1. 依赖确认（Pom.xml） #

Spring Boot 3需使用Jakarta版本依赖，Qpid JMS 2.2.0已适配Jakarta，避免引入冗余依赖引发冲突：

<dependencies>
    <!-- Spring Boot JMS Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jms</artifactId>
    </dependency>
    <!-- Qpid JMS AMQP 1.0 Client -->
    <dependency>
        <groupId>org.apache.qpid</groupId>
        <artifactId>qpid-jms-client</artifactId>
        <version>2.2.0</version>
    </dependency>
</dependencies>

2. 生产者实现（发送具体主题消息） #

生产者无需特殊配置，只需发送到具体层级的主题，消费者通过通配符订阅匹配消息：

import jakarta.jms.JmsTemplate;
import org.springframework.stereotype.Component;

@Component
public class AmqpProducer {
    private final JmsTemplate jmsTemplate;

    public AmqpProducer(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    // 发送到具体主题，如TEST.ORDER、TEST.PAYMENT
    public void sendTopicMessage(String specificTopic, String content) {
        jmsTemplate.convertAndSend("topic:" + specificTopic, content);
    }
}

3. 消费者实现（通配符订阅） #

通过@JmsListener注解指定带通配符的主题地址，实现批量订阅：

import jakarta.jms.Message;
import jakarta.jms.TextMessage;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class AmqpConsumer {

    // 订阅TEST下所有单层级主题
    @JmsListener(destination = "topic:TEST.*")
    public void handleSingleLevelTestMessages(Message message) throws Exception {
        if (message instanceof TextMessage textMsg) {
            System.out.println("Received single-level test message: " + textMsg.getText());
        }
    }

    // 订阅TEST下所有层级的主题
    @JmsListener(destination = "topic:TEST.>")
    public void handleAllLevelTestMessages(Message message) throws Exception {
        if (message instanceof TextMessage textMsg) {
            System.out.println("Received all-level test message: " + textMsg.getText());
        }
    }
}

4. Spring Boot配置（application.yml） #

配置AMQP连接信息，启用主题模式：

spring:
  jms:
    broker-url: amqp://localhost:5672  # ActiveMQ默认AMQP端口
    username: admin
    password: admin
    template:
      pub-sub-domain: true  # 启用主题订阅模式（默认是队列）

若ActiveMQ未开启AMQP协议，需在activemq.xml中添加连接器：

<transportConnectors>
    <transportConnector name="openwire" uri="tcp://0.0.0.0:61616"/>
    <transportConnector name="amqp" uri="amqp://0.0.0.0:5672"/>
</transportConnectors>

关键注意事项 #

• 生产者发送具体主题，消费者用通配符订阅，这是实现多消费者接收相同消息的核心逻辑，替代VirtualTopic的Consumer.*.VirtualTopic.XXX写法。
• 不要同时引入qpid-amqp-1-0-client-jms，Qpid JMS 2.x已整合AMQP 1.0实现，冗余依赖会导致类加载冲突。
• 确保ActiveMQ用户拥有对应主题的读写权限，避免消息发送/订阅失败。

内容的提问来源于stack exchange，提问作者Mançaux Pierre-Alexandre