Spring Boot 2.7+Spring JMS多应用同时消费同一Topic的实现问题

你碰到的是JMS Topic订阅的典型配置问题：默认设置下，Spring JMS会将Topic当作Queue处理，或是多个消费者应用共享了同一订阅身份，导致消息被轮询分发而非广播给所有订阅者。以下是具体的修复步骤：

一、修正发布端：确保消息发送到Topic #

Spring JMS的JmsTemplate默认pubSubDomain为false（默认适配Queue），必须显式配置使其支持Topic模式：

方法1：全局配置JmsTemplate为Topic模式 #

在发布者应用的配置类中自定义JmsTemplate：

@Configuration
public class JmsConfig {
    @Bean
    public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
        JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
        jmsTemplate.setPubSubDomain(true); // 启用Topic广播模式
        return jmsTemplate;
    }
}

方法2：发送时指定Topic目的地 #

如果不想全局修改JmsTemplate行为，可在发送时直接指定目的地类型为Topic：

public void publish(final String name) {
    jmsTemplate.convertAndSend(Session.TOPIC, "name-topic", name);
}

二、修正消费端：保证每个应用是独立订阅者 #

JMS Topic的广播特性要求每个消费者应用拥有唯一的订阅身份，否则会被视为同一订阅者的多实例，消息会被轮询分配。

1. 非持久订阅（临时订阅） #

对于临时订阅，需确保每个消费者应用的连接使用唯一的clientId，并配置监听器容器适配Topic：

步骤1：添加YAML配置

在消费者A和B的application.yml中分别配置：

spring:
  jms:
    listener:
      topic:
        concurrency: 1
    template:
      pub-sub-domain: true
  activemq: # 若使用其他JMS代理，替换为对应配置项
    client-id: ${spring.application.name}-unique-client # 用应用名保证clientId唯一

步骤2：创建Topic监听器容器工厂

在每个消费者应用的配置类中添加：

@Configuration
public class JmsListenerConfig {
    @Bean
    public DefaultJmsListenerContainerFactory jmsTopicListenerContainerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(true); // 启用Topic模式
        return factory;
    }
}

步骤3：更新@JmsListener注解

指定使用Topic容器工厂：

// 消费者A
@JmsListener(destination = "name-topic", containerFactory = "jmsTopicListenerContainerFactory")
void welcome(final String name) {
    System.out.println("Hello, " + name);
}

// 消费者B
@JmsListener(destination = "name-topic", containerFactory = "jmsTopicListenerContainerFactory")
void welcome(final String name) {
    System.out.println("Good bye, " + name);
}

2. 持久订阅（推荐，避免离线丢消息） #

如果需要保证消费者离线时消息不丢失，需使用持久订阅，此时必须为每个消费者指定唯一的clientId和subscriptionName：

消费者A配置

@JmsListener(destination = "name-topic", 
             containerFactory = "jmsTopicListenerContainerFactory",
             subscription = "consumer-a-sub")
void welcome(final String name) {
    System.out.println("Hello, " + name);
}

消费者B配置

@JmsListener(destination = "name-topic", 
             containerFactory = "jmsTopicListenerContainerFactory",
             subscription = "consumer-b-sub")
void welcome(final String name) {
    System.out.println("Good bye, " + name);
}

对应YAML配置

消费者A的application.yml：

spring:
  activemq:
    client-id: consumer-a-client # 唯一clientId

消费者B的application.yml：

spring:
  activemq:
    client-id: consumer-b-client # 唯一clientId

三、验证效果 #

重启三个应用后，发布者发送John和Jane：

• 消费者A输出：

  Hello, John
  Hello, Jane
  

• 消费者B输出：

  Good bye, John
  Good bye, Jane
  

完全符合每条消息被所有消费者接收的预期。

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