springboot整合activemq，实现queue，topic同时支持
1、JMS中定义了两种消息模型：点对点（point to point， queue）和发布/订阅（publish/subscribe，topic），区别如下：
（1）queue（点对点）：不可重复消费。消息生产者发送消息到queue，然后消息消费者从中取出并消费消息；消息消费后将不再存储在queue中；queue支持存在多个消费者，但是对于一条消息来说，只能有一个消费者进行消费；当消息没有消费者的时候，消息将一直保存，知道有消费者消费。
（2）topic（发布/订阅）：可以重复消费。消息生产者发布消息到topic，同时有多个订阅者订阅该消息；发布到topic中的消息会被所有的订阅者消费；当生产者发布消息，不管是否有订阅者，消息都不会保存。
2、之前的笔记中对p2p的消息发布进行了测试，下面使用发布/订阅模式来进行消息发布与消费，首先要增加配置，开启发布/订阅模式：spring.jms.pub-sub-domain=true
3、登陆activemq控制台，新建topic：test.topic
4、在Application中新增代码，方便注入topic
@Bean
public Topic topic() {
return new ActiveMQTopic("test.topic");
}
5、添加消息发布接口及实现
/**
*
*/
package com.zc.app.test.service; import javax.jms.Destination; /**
* topic 消息发布接口
*
*/
public interface PublishService { public void publish(String msg) ; public void publish(Destination des,String msg);
}
PublishService
/**
*
*/
package com.zc.app.test.service.impl; import javax.jms.Destination;
import javax.jms.Topic; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service; import com.zc.app.test.service.PublishService; /**
* 消息发布实现
*
*/
@Service
public class PublishServiceImpl implements PublishService{ @Autowired
private Topic topic; @Autowired
private JmsMessagingTemplate jms; //发送消息
@Override
public void publish(String msg) {
jms.convertAndSend(topic, msg); }
//发送消息到指定的topic
@Override
public void publish(Destination des, String msg) {
jms.convertAndSend(des, msg);
} }
PublishServiceImpl
6、增加消息订阅者
/**
*
*/
package com.zc.app.test.jms; import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component; /**
* 消息消费者
*
*/
@Component
public class TopicSubscribe { @JmsListener(destination="test.topic")
public void ubscribe1(String msg) {
System.out.println("消息订阅者1："+msg);
} @JmsListener(destination="test.topic")
public void ubscribe2(String msg) {
System.out.println("消息订阅者2："+msg);
} @JmsListener(destination="test.topic")
public void subscribe3(String msg) {
System.out.println("消息订阅者3："+msg);
}
}
TopicSubscribe
7、contoller中增加发送消息接口
@Autowired
private PublishService publishService; @GetMapping("topic")
public String topic(String msg) {
publishService.publish(msg);
return "send topic msg";
}
8、访问http://localhost:8080/msg/topic?msg=12468487，控制台输出，说明订阅成功
9、这时发现向queue中写入数据后，消息消费者没有输出了，这时因为activemq默认只支持点对点模式，更改配置文件后又仅支持发布/订阅模式。要同时支持两种模式，就需要为topic定义独立的JmsListenerContainer
10、在Application中增加代码
@Bean
public JmsListenerContainerFactory<?> jmsListenerContainerTopic() {
DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
bean.setPubSubDomain(true);
bean.setConnectionFactory(new ActiveMQConnectionFactory());
return bean;
}
11、在TopicSubscribe的@JmsListener注解中增加containerFactory设置，去掉配置文件中的spring.jms.pub-sub-domain=true，这时就可以实现两种模式同时使用了，我们更改一下TopicSubscribe
/**
*
*/
package com.zc.app.test.jms; import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component; /**
* 订阅者
*
*/
@Component
public class TopicSubscribe { @JmsListener(destination="test.topic",containerFactory="jmsListenerContainerTopic")
public void ubscribe1(String msg) {
System.out.println("消息订阅者1："+msg);
} @JmsListener(destination="test.topic",containerFactory="jmsListenerContainerTopic")
public void ubscribe2(String msg) {
System.out.println("消息订阅者2："+msg);
} @JmsListener(destination="test.topic")
public void subscribe3(String msg) {
System.out.println("消息订阅者3："+msg);
}
}
TopicSubscribe
12、再次访问http://localhost:8080/msg/topic?msg=12468487，发现只有订阅者1、2输出了消息，同时queue的消费者也可以收到消息了。