如何在Red Hat AMQ 7中用AMQP 1.0配置主题并在Spring Boot消费？

问题分析 #

1. 生产者消息未在控制台统计：你直接发送消息到多播队列名称而非多播地址，且JmsTemplate默认使用点对点模式，消息未经过主题地址路由，导致控制台地址消息数无增长。
2. 消费者启动报错：jms.topic.demoTopic地址未配置队列支持，本质是未正确配置JMS监听的发布订阅模式，且目标地址与AMQ控制台创建的多播地址不匹配。

解决方案步骤 #

1. 确认AMQ控制台配置 #

确保你创建的是多播（Multicast）类型地址（而非点对点Anycast），地址名称记为demoTopic（示例），多播队列可手动创建或由AMQ自动为消费者生成。

2. 修改Spring Boot配置文件 (application.properties) #

amqphub.amqp10jms.remote-url=amqp://localhost:61616
amqphub.amqp10jms.username=admin
amqphub.amqp10jms.password=admin
# 指定AMQ控制台创建的多播地址名称
my.topic.name=demoTopic
# 全局配置JMS为发布订阅模式
spring.jms.pub-sub-domain=true

3. 调整生产者代码 #

确保发送目标为多播地址，且JmsTemplate启用发布订阅模式：

@Component
public class Producer {

    private static final Logger LOG = LoggerFactory.getLogger(Producer.class);

    @Autowired
    private JmsTemplate jmsTemplate;

    @Value("${my.topic.name}")
    private String topicName;

    @Scheduled(fixedDelay = 2000L)
    public void sendMessages() {
        sendMessage("Hello World topic");
    }

    public void sendMessage(final String payload) {
        LOG.info("============= Sending: " + payload);
        // 明确指定发送到Topic类型目标
        this.jmsTemplate.convertAndSend(new Topic(topicName), payload);     
    }
}

也可通过@Bean显式配置JmsTemplate：

@Bean
public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
    JmsTemplate template = new JmsTemplate(connectionFactory);
    template.setPubSubDomain(true); // 强制开启发布订阅模式
    return template;
}

4. 修复消费者代码并配置监听工厂 #

创建支持发布订阅的JMS监听工厂，并指定正确的目标地址：

@Configuration
public class JmsConfig {

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(true); // 配置为发布订阅模式
        return factory;
    }
}

消费者类修改：

@Component
public class Consumer {

    private static final Logger LOG = LoggerFactory.getLogger(Consumer.class);

    // 绑定配置好的发布订阅工厂，目标地址使用配置文件中的多播地址
    @JmsListener(destination = "${my.topic.name}", containerFactory = "jmsListenerContainerFactory")
    public void processMsg(final String message) {
        LOG.info("Received: " + message);
    }
}

关键说明 #

• AMQP 1.0在Red Hat AMQ中，主题（发布订阅）对应多播地址，点对点对应点对点地址，不要混淆地址和队列名称。
• JMS规范中，Topic对应发布订阅，Queue对应点对点，需确保JmsTemplate和JmsListenerContainerFactory的pubSubDomain属性正确设置为true。
• 若消费者需要持久化订阅，可在监听工厂中配置客户端ID和持久化订阅名称，确保消费者重启后能接收离线消息。

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