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); }
/** * */ 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); } }
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); } }
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); } }
12、再次访问http://localhost:8080/msg/topic?msg=12468487,发现只有订阅者1、2输出了消息,同时queue的消费者也可以收到消息了。
原文:https://www.cnblogs.com/codingstudy/p/10303993.html