SpringBoot(二十二)集成RabbitMQ---MQ实战演练
Posted 请叫我头头哥
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了SpringBoot(二十二)集成RabbitMQ---MQ实战演练相关的知识,希望对你有一定的参考价值。
RabbitMQ是一个在AMQP基础上完成的,可复用的企业消息系统。他遵循Mozilla Public License开源协议。RabbitMQ是流行的开源消息队列系统,用erlang语言开发。RabbitMQ是AMQP(高级消息队列协议)的标准实现。
消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.
如果你还没有安装rabbitmq的,可以看看这篇《centos安装MQ》
不说了不说了,来一张图直截了当的看看MQ工作的具体过程:
开局一张图 故事全靠编.从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息队列,并且当消息队列收到消息之后,接收消息队列传来的消息,并且给予相应的处理.消息队列常用于分布式系统之间互相信息的传递.
v基础概念
对于RabbitMQ来说,除了这三个基本模块以外,还添加了一个模块,即交换机(Exchange).它使得生产者和消息队列之间产生了隔离,生产者将消息发送给交换机,而交换机则根据调度策略把相应的消息转发给对应的消息队列.那么RabitMQ的工作流程如下所示:
关于rabbitmq几个基础名词的介绍:
交换机的主要作用是接收相应的消息并且绑定到指定的队列.交换机有四种类型,分别为Direct,topic,headers,Fanout:
v实战演练
注:若是现有工程引入MQ,则添加Maven引用。
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
这里我们延续之前springboot系列博文中的例子hellospringboot,在已有项目中添加mq的Maven引用。
在application.properties文件当中引入RabbitMQ基本的配置信息
# ----- MQ -------- #
spring.rabbitmq.host=192.168.11.108
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
package com.demo.mq.model; import java.io.Serializable; import java.util.UUID; /** * Created by toutou on 2019/1/1. */ public class MyModel implements Serializable { private static final long serialVersionUID = 1L; private UUID id; private String info; public UUID getId() { return id; } public void setId(UUID id) { this.id = id; } public String getInfo() { return info; } public void setInfo(String info) { this.info = info; } }
package com.demo.mq.common; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Scope; /** * Created by toutou on 2019/1/1. */ @Configuration public class RabbitConfig { @Value("${spring.rabbitmq.host}") private String host; @Value("${spring.rabbitmq.port}") private int port; @Value("${spring.rabbitmq.username}") private String username; @Value("${spring.rabbitmq.password}") private String password; public static final String EXCHANGE_A = "my-mq-exchange_A"; public static final String EXCHANGE_B = "my-mq-exchange_B"; public static final String QUEUE_A = "QUEUE_A"; public static final String QUEUE_B = "QUEUE_B"; public static final String ROUTINGKEY_A = "spring-boot-routingKey_A"; public static final String ROUTINGKEY_B = "spring-boot-routingKey_B"; @Bean public CachingConnectionFactory connectionFactory() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port); connectionFactory.setUsername(username); connectionFactory.setPassword(password); connectionFactory.setVirtualHost("/"); connectionFactory.setPublisherConfirms(true); return connectionFactory; } @Bean @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) public RabbitTemplate rabbitTemplate() { RabbitTemplate template = new RabbitTemplate(connectionFactory()); return template; } /** * 针对消费者配置 * 1. 设置交换机类型 * 2. 将队列绑定到交换机 FanoutExchange: 将消息分发到所有的绑定队列,无routingkey的概念 HeadersExchange :通过添加属性key-value匹配 DirectExchange:按照routingkey分发到指定队列 TopicExchange:多关键字匹配 */ @Bean public DirectExchange defaultExchange() { return new DirectExchange(EXCHANGE_A); } /** * 获取队列A * @return */ @Bean public Queue queueA() { return new Queue(QUEUE_A, true); //队列持久 } /** * 获取队列B * @return */ @Bean public Queue queueB() { return new Queue(QUEUE_B, true); //队列持久 } /** * 把交换机,队列,通过路由关键字进行绑定 * @return */ @Bean public Binding binding() { return BindingBuilder.bind(queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A); } /** * 一个交换机可以绑定多个消息队列,也就是消息通过一个交换机,可以分发到不同的队列当中去。 * @return */ @Bean public Binding bindingB(){ return BindingBuilder.bind(queueB()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_B); } }
package com.demo.mq.producer; import com.demo.mq.common.RabbitConfig; import com.demo.mq.model.MyModel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; /** * Created by toutou on 2019/1/1. */ @Component public class MyProducer implements RabbitTemplate.ConfirmCallback { private final Logger logger = LoggerFactory.getLogger(this.getClass()); //由于rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE,所以不能自动注入 private RabbitTemplate rabbitTemplate; /** * 构造方法注入rabbitTemplate */ @Autowired public MyProducer(RabbitTemplate rabbitTemplate) { this.rabbitTemplate = rabbitTemplate; rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果为单例的话,那回调就是最后设置的内容 } public void sendMsg(MyModel model) { //把消息放入ROUTINGKEY_A对应的队列当中去,对应的是队列A rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A, RabbitConfig.ROUTINGKEY_A, model); } /** * 回调 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { logger.info(" 回调id:" + correlationData); if (ack) { logger.info("消息成功消费"); } else { logger.info("消息消费失败:" + cause); } } }
package com.demo.mq.receiver; import com.demo.mq.common.RabbitConfig; import com.demo.mq.model.MyModel; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * Created by toutou on 2019/1/1. */ @Component @RabbitListener(queues = RabbitConfig.QUEUE_A) public class MyReceiver { @RabbitHandler public void process(MyModel model) { System.out.println("接收处理队列A当中的消息: " + model.getInfo()); } }
package com.demo.controller; import com.demo.mq.model.MyModel; import com.demo.mq.producer.MyProducer; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import java.util.UUID; /** * Created by toutou on 2019/1/1. */ @RestController @Slf4j public class MyMQController { @Autowired MyProducer myProducers; @GetMapping("/mq/producer") public String myProducer(String content){ MyModel model = new MyModel(); model.setId(UUID.randomUUID()); model.setInfo(content); myProducers.sendMsg(model); return "已发送:" + content; } }
2.9.1 在页面中请求http://localhost:8081/mq/producer?content=hello rabbitmq
2.9.2 查看http://ip:15672/#/queues的变化
关于RabbitMQ Management有疑问的,可以看上篇博文。《浅谈RabbitMQ Management》。
2.9.3 查看消费者日志记录
这样一个完整的rabbitmq实例就有了。
v源码地址
https://github.com/toutouge/javademosecond/tree/master/hellospringboot
作 者:请叫我头头哥
出 处:http://www.cnblogs.com/toutou/
关于作者:专注于基础平台的项目开发。如有问题或建议,请多多赐教!
版权声明:本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文链接。
特此声明:所有评论和私信都会在第一时间回复。也欢迎园子的大大们指正错误,共同进步。或者直接私信我
声援博主:如果您觉得文章对您有帮助,可以点击文章右下角【推荐】一下。您的鼓励是作者坚持原创和持续写作的最大动力!
以上是关于SpringBoot(二十二)集成RabbitMQ---MQ实战演练的主要内容,如果未能解决你的问题,请参考以下文章
第二十二篇Flowable之SpringBoot集成FlowableUI
SpringBoot和Vue集成视频播放组件——基于SpringBoot和Vue的后台管理系统项目系列博客(二十二)
springboot系列(二十二):集成easypoi实现Excel文件的导入导出(准备篇)
springboot系列(二十二):集成easypoi实现Excel文件的导入导出(准备篇)