SpringBoot集成AMQ消息队列

首先新建一个SpringBoot项目，下面开始正式的集成操作。

第一步：

打开pom.xml文件，添加依赖。这里等待maven仓库加载完成。

org.springframework.bootspring-boot-starter-activemqorg.springframework.bootspring-boot-starter-weborg.messaginghubpooled-jms

第二步：

打开application.properties将消息队列相关参数进行配置

#消息队列配置
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.packages.trust-all=true
spring.activemq.in-memory=true
spring.activemq.pool.enabled=false
spring.activemq.user="admin"
spring.activemq.password="admin"
#如果为true，则是Topic；如果是false或者默认，则是queue，开启了双向队列
spring.jms.pub-sub-domain=true

第三步：

创建消息队列服务，这里需要注意，我创建了两个消息队列发送，分别为queue和topic，它们在第二步配置文件中有体现，他俩的区别如下：

A.Queue(点对点，不可重复消费)
1.生产者生产消息发送到queue中，然后消费者从queue中取出并且消费消息。
2.消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息。
3.Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费、其它的则不能消费此消息了。 当消费者不存在时，消息会一直保存，直到有消费消费
B.Topic(发布/订阅：Topic，可以重复消费)
1.消息生产者（发布）将消息发布到topic中，同时有多个消费者（订阅）消费该消息。
2.和queue方式不同，发布到topic的消息会被所有订阅者消费。
3.当生产者发布消息，不管是否有消费者。都不会保存消息。
4.如果没有消费者，生产的消息就是废消息。也就是说topic必须有消费者才有意义。
5.后订阅的消费者收不到之前发布的消息。

@Component
public class QueueMessageService {@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;/*** 使用queue模式注册队列** @param queue* @param msg*/public void setQueue(String queue, String msg) {jmsMessagingTemplate.convertAndSend(new ActiveMQQueue(queue), msg);}/*** 使用topic模式注册队列** @param topic* @param msg*/public void setTopic(String topic, String msg) {jmsMessagingTemplate.convertAndSend(new ActiveMQTopic(topic), msg);}
}


第四步：

创建消息生产者

@RestController
@RequestMapping("message")
public class ProducerController {@Autowiredprivate QueueMessageService queueMessageService;/*** 消息生产者*/@GetMapping("sendmsg")public void sendmsg(String msg) {// 指定消息发送的目的地及内容queueMessageService.setQueue("queue", msg);}@GetMapping("sendmsgx")public void sendmsgtox(String key, String msg) {// 由生产者指定消息发送的目的地及内容queueMessageService.setQueue(key, msg);}@GetMapping("send")public void send(String msg) {// 指定消息发送的目的地及内容//System.out.println(new Date().toString());queueMessageService.setTopic("topic", msg);}}


第五步：

创建消息消费者

@RestController
public class ConsumerController {/*** 监听队列** @param message*/@JmsListener(destination = "queue")public void readActiveQueue(String message) {System.out.println("queue模式接收1：" + message);//TODO something}@JmsListener(destination = "queue_b")public void readQueue(String message) {System.out.println("queue模式接收2：" + message);}@JmsListener(destination = "queue_byte")public void readQueueByte(byte[] message) {System.out.println("queue模式接收3：" + message);}@JmsListener(destination = "topic")public void receiveMessage(String message) throws InterruptedException {System.out.println(new Date().toString() + "topic1接收:" + message);}@JmsListener(destination = "topic")public void readResponse(String message) throws InterruptedException {System.out.println(new Date().toString() + "topic2接收:" + message);}
}


测试

1.测试Topic，在浏览器输入http://127.0.0.1:8080/message/send?msg=test--topic

这时我们发现两个topic都接收到了请求，测试成功！

2.测试Queue，首先我们要先在配置文件中关闭，然后在浏览器输入http://127.0.0.1:8080/message/sendmsg?msg=test--queue

#spring.jms.pub-sub-domain=true

 这时我们在控制台中接收到了消息队列，测试成功！