求解决：ActiveMQ在发布订阅模式下有多个consumer，consumer会递增式的重复接收消息
问题：springboot集成ActiveMQ实现了消费队列和发布订阅两种消息类型，发布订阅模式下有TopicConsumer1和TopicConsumer2，监听的destination都是alarm.topic，但这两个consumer会递增式的重复接收消息。我的实现代码如下：
test入口类
@RunWith(SpringRunner.class)
@SpringBootTest(classes = MessageCenterApplication.class)
public class ActivemqTest {
@Autowired
private Producer producer;
@Test
public void testSendTopicMessage(){
for (int i = 0; i < 5; i++) {
producer.sendTopicMessage(Destinations.ALARM_TOPIC, "here is a topic message, the number is " + i);
}
}
/*输出结果如下，这个输出结果不是我所期望的
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 0
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 0
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 1
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 1
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 1
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 1
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 2
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 2
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 2
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 2
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 2
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 2
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 3
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 3
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 3
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 3
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 3
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 3
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 3
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 3
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 4
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 4
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 4
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 4
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 4
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 4
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 4
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 4
TopicConsumer2接收到的消息内容为：here is a topic message, the number is 4
TopicConsumer1接收到的消息内容为：here is a topic message, the number is 4*/
}
Producer实现类
@Component
public class Producer {
@Autowired
@Qualifier("jmsQueueMessagingTemplate")
private JmsMessagingTemplate jmsQueueMessagingTemplate;
@Autowired
@Qualifier("jmsTopicMessagingTemplate")
private JmsMessagingTemplate jmsTopicMessagingTemplate;
/**
* 使用queue消息类型（点对点）
/
public void sendQueueMessage(String destination, final String message){
Queue queue = new ActiveMQQueue(destination);
jmsQueueMessagingTemplate.convertAndSend(queue, message);
}
/*
* 使用topic消息类型
*/
public void sendTopicMessage(String destination, final String message){
Topic topic = new ActiveMQTopic(destination);
jmsTopicMessagingTemplate.convertAndSend(topic, message);
}
}
TopicConsumer1和TopicConsumer2的实现代码
@Component
public class TopicConsumer1 {
@JmsListener(destination = Destinations.ALARM_TOPIC, containerFactory = "jmsListenerContainerFactory4Topic")
public void receive(String message){
System.out.println("TopicConsumer1接收到的消息内容为：" + message);
}
}
@Component
public class TopicConsumer2 {
@JmsListener(destination = Destinations.ALARM_TOPIC, containerFactory = "jmsListenerContainerFactory4Topic")
public void receive(String message){
System.out.println("TopicConsumer2接收到的消息内容为：" + message);
}
}
jmsTopicMessagingTemplate对应的bean的配置
@Configuration
public class JmsMessagingConfig {
@Autowired
@Qualifier(value = "jmsQueueTemplate")
private JmsTemplate jmsQueueTemplate;
@Autowired
@Qualifier(value = "jmsTopicTemplate")
private JmsTemplate jmsTopicTemplate;
@Bean(name = "jmsQueueMessagingTemplate")
public JmsMessagingTemplate jmsQueueMessagingTemplate(){
return new JmsMessagingTemplate(jmsQueueTemplate);
}
@Bean(name = "jmsTopicMessagingTemplate")
public JmsMessagingTemplate jmsTopicMessagingTemplate(){
return new JmsMessagingTemplate(jmsTopicTemplate);
}
}
jmsTopicTemplate对应的bean的配置
@Configuration
public class JmsConfig {
private final JmsProperties jmsProperties;
private final ObjectProvider<DestinationResolver> destinationResolver;
private final ObjectProvider<MessageConverter> messageConverter;
public JmsConfig(JmsProperties jmsProperties, ObjectProvider<DestinationResolver> destinationResolver,
ObjectProvider<MessageConverter> messageConverter) {
this.jmsProperties = jmsProperties;
this.destinationResolver = destinationResolver;
this.messageConverter = messageConverter;
}
@ConfigurationProperties(prefix = "spring.activemq")
@Bean(name = "jmsQueueTemplate")
@Primary
public JmsTemplate jmsQueueTemplate(PooledConnectionFactory pooledConnectionFactory) {
JmsTemplate template = new JmsTemplate(pooledConnectionFactory);
// 设置为使用queue消息类型
template.setPubSubDomain(false);
DestinationResolver destinationResolver = (DestinationResolver) this.destinationResolver.getIfUnique();
if (destinationResolver != null) {
template.setDestinationResolver(destinationResolver);
}
MessageConverter messageConverter = (MessageConverter) this.messageConverter.getIfUnique();
if (messageConverter != null) {
template.setMessageConverter(messageConverter);
}
// deliveryMode, priority, timeToLive 的开关，要生效，必须配置为true，默认false
template.setExplicitQosEnabled(true);
// DeliveryMode.NON_PERSISTENT=1:非持久 ; DeliveryMode.PERSISTENT=2:持久
template.setDeliveryMode(DeliveryMode.PERSISTENT);
// 默认不开启事务
// 如果不启用事务，则会导致XA事务失效；
// 作为生产者如果需要支持事务，则需要配置SessionTransacted为true
template.setSessionTransacted(true);
return template;
}
@ConfigurationProperties(prefix = "spring.activemq")
@Bean(name = "jmsTopicTemplate")
public JmsTemplate jmsTopicTemplate(PooledConnectionFactory pooledConnectionFactory) {
JmsTemplate template = new JmsTemplate(pooledConnectionFactory);
// 设置为使用topic消息类型
template.setPubSubDomain(true);
DestinationResolver destinationResolver = (DestinationResolver) this.destinationResolver.getIfUnique();
if (destinationResolver != null) {
template.setDestinationResolver(destinationResolver);
}
MessageConverter messageConverter = (MessageConverter) this.messageConverter.getIfUnique();
if (messageConverter != null) {
template.setMessageConverter(messageConverter);
}
// deliveryMode, priority, timeToLive 的开关，要生效，必须配置为true，默认false
template.setExplicitQosEnabled(true);
// DeliveryMode.NON_PERSISTENT=1:非持久 ; DeliveryMode.PERSISTENT=2:持久
template.setDeliveryMode(DeliveryMode.PERSISTENT);
// 默认不开启事务
// 如果不启用事务，则会导致XA事务失效；
// 作为生产者如果需要支持事务，则需要配置SessionTransacted为true
template.setSessionTransacted(true);
return template;
}
@ConfigurationProperties(prefix = "spring.activemq")
@Bean(name = "jmsListenerContainerFactory4Topic")
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory4Topic(PooledConnectionFactory pooledConnectionFactory) {
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
factory.setConnectionFactory(pooledConnectionFactory);
// 设置为使用topic消息类型
factory.setPubSubDomain(true);
//如果单独配置了分布式事务，则启用
// if (this.transactionManager != null) {
// factory.setTransactionManager(transactionManager);
// } else {
factory.setSessionTransacted(Boolean.valueOf(true));
// }
JmsProperties.Listener listener = jmsProperties.getListener();
factory.setAutoStartup(listener.isAutoStartup());
if (listener.getAcknowledgeMode() != null) {
factory.setSessionAcknowledgeMode(Integer.valueOf(listener.getAcknowledgeMode().getMode()));
}
String concurrency = listener.formatConcurrency();
if (concurrency != null){
factory.setConcurrency(concurrency);
}
return factory;
}
@ConfigurationProperties(prefix = "spring.activemq")
@Bean(name = "jmsListenerContainerFactory4Queue")
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory4Queue(PooledConnectionFactory pooledConnectionFactory) {
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
factory.setConnectionFactory(pooledConnectionFactory);
// 设置为使用queue消息类型
factory.setPubSubDomain(false);
//如果单独配置了分布式事务，则启用
// if (this.transactionManager != null) {
// factory.setTransactionManager(transactionManager);
// } else {
factory.setSessionTransacted(Boolean.valueOf(true));
// }
JmsProperties.Listener listener = jmsProperties.getListener();
factory.setAutoStartup(listener.isAutoStartup());
if (listener.getAcknowledgeMode() != null) {
factory.setSessionAcknowledgeMode(Integer.valueOf(listener.getAcknowledgeMode().getMode()));
}
String concurrency = listener.formatConcurrency();
if (concurrency != null){
factory.setConcurrency(concurrency);
}
return factory;
}
}
yml配置文件的配置
spring:
#jms配置，配合activemq一起配置
jms:
listener:
concurrency: 1
max-concurrency: 50
#activemq配置
activemq:
#默认使用spring-boot-starter-activemq内嵌的activemq，除非明确指定了broker-url
in-memory: false
user: system
password: manager
broker-url: auto+nio://192.168.137.5:61616
pool:
enabled: true
max-connections: 500
