MQ在工作中用途还是比较多的，RabbitMQ又是比较容易上手并且在企业中用的比较多的一种消息服务，本篇文章借鉴于ginobefun的文章和ConanLi的文章，一方面是加深理解，一方面也是补充自己在MQ的不足。
一、AMQP基础
Paste_Image.png
二、RabbitMQ
基础概念
- 生产者、消费者
- 队列
- 交换器
- 绑定
生产者、消费者
- 生产者（producer）创建消息，然后发送到代理服务器（RabbitMQ）
- 消费者（consumer）连接到代理服务器上，并订阅到队列（queue）上；当消费者接收到消息时，它只得到消息的一部分：有效载荷（标签并没有随有效载荷一同传递）
- 信道（channel）建立在”真实的”TCP连接内的虚拟连接；不论是发布信息、订阅队列或是接收消息，都是通过信道完成的；不使用TCP连接主要是因为对于操作系统而言建立和销毁TCP会话非常昂贵的开销；在一条TCP连接上创建多少条信道是没有限制的
- 消息包含两部分：有效载荷（payload）和标签（label）；有效载荷就是你想要传输的数据（可以是任何格式的任何内容）；标签描述了有效载荷，并且RabbitMQ用它来决定谁将获得消息的拷贝（之后举例说明）
队列(queue)
Paste_Image.png
- Queue（队列）是RabbitMQ的内部对象，用于存储消息
- 主体流程：
- 队列类似一个broker角色，生产者将内容（消息）发送到队列
- 队列进行存储，消费者将消息消费
- 消费者确认消费消息（ack）
- 生产者和消费者都可以通过来创建队列：
channel.queueDeclare(QUEUE_NAME, durable, exclusive, autoDelete, arguments);
- durable：队列名称，不指定则随机生成
- exclusive：设置为true则为私有队列，只有当前消费者可以订阅；
- autoDelete：设置为true时最后一个消费者取消订阅将自动移除队列；
- arguments：参数
交换器&绑定
Paste_Image.png
RabbitMQ的消息不是直接从生产者发送到队列的，而是要经过交换器然后才可以到达队列：
- 生成者把消息发布到交换器上；
- 消息最终到达队列，并被消费者接收；
- 绑定决定了消息如何从交换器到特定的队列；
四种交换器类型：
- fanout：把所有发送到该Exchange的消息路由到所有与它绑定的Queue中
- direct：把消息路由到bindingKey与routingKey完全匹配的Queue中
- topic：把消息路由到bindingKey与routingKey模糊匹配的Queue中
- headers：headers类型的Exchange不依赖于routingKey与bindingKey的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配
1、fanout
Paste_Image.png
- 生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。
2、direct
Paste_Image.png
- routingKey=”error”发送消息，则会同时路由到Queue1（amqp.gen-S9b…）和Queue2（amqp.gen-Agl…）
- routingKey=”info”或routingKey=”warning”发送消息，则只会路由到Queue2
- 以其他routingKey发送消息，则不会路由到这两个Queue中
3、topic
Paste_Image.png
- routingKey=”quick.orange.rabbit”发送信息，则会同时路由到Q1与Q2
- routingKey=”lazy.orange.fox”发送信息，则只会路由到Q1
- routingKey=”lazy.brown.fox”发送消息，则只会路由到Q2
- routingKey=”lazy.pink.rabbit”发送消息，则只会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）
- routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”发送消息，则会被丢弃，它们并没有匹配任何bindingKey
4、header
headers类型的Exchange不依赖于routingKey与bindingKey的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。
三、SpringBoot+RabbitMQ实战
** Docker环境下安装RabbitMQ**
# 下载rabbitmq的docker镜像和managerment的镜像
docker pull rabbitmq:management
# 启动rabbitmq镜像
docker run -d --name rabbitmq --publish 5671:5671 \
--publish 5672:5672 --publish 4369:4369 --publish 25672:25672 --publish 15671:15671 --publish 15672:15672 \
rabbitmq:management
-
端口解释：
- 4369:epmd(Erlang Port Mapper Daemon)
- 25672:Erlang distribution
- 5672, 5671:AMQP 0-9-1 without and with TLS
- 15672:if management plugin is enabled
- 61613, 61614:if STOMP is enabled
- 1883, 8883:if MQTT is enabled
-
默认访问路径：
http://localhost:15672 -
默认用户名和密码：
guest/guest
介于代码粘贴进来比较多，提供项目GitHub地址，本文提取代码片段进行讲解
项目结构
Paste_Image.png
关键参数
- @RabbitListener(queues = "xxx")
- RabbitTemplate
- TopicExchange
- Binding
- Queue
代码段讲解
- 创建队列
@Bean
public Queue ssdQueue(){
return new Queue("hello");
}
- 创建topic类型交换器
@Bean
public TopicExchange topicExchange() {
return new TopicExchange("topicExchange");
}
- 创建fanout类型的交换器
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");
}
- 绑定topic类型的交换器
@Bean
public Binding bindingExchangeTopicA(Queue topicAQueue, TopicExchange topicExchange) {
return BindingBuilder.bind(topicAQueue).to(topicExchange).with("topic.a");
}
- 绑定fanout类型的交换器
@Bean
public Binding bindingExchangeFanoutC(Queue fanoutCQueue, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(fanoutCQueue).to(fanoutExchange);
}
四、总结
本片文章主要讲解了以下RabbitMQ的一些基础概念和使用SpringBoot和RabbitMQ整合的几个案例，也是笔者结合博客和官网写的一片总结，总结的也比较基础，没有包括一些高级的内容，例如事务，最终一致性，重复消息和顺序消息的处理等等。
网友评论