[消息队列] RabbitMQ

标签：server   流量   工作   mq服务器   format   请求   return   区别   异步处理   

介绍

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件。主要用来做异步处理、应用解耦、流量削峰、日志处理等等方面的处理。

特点

可靠性: RabbitMQ使用一些机制来保证可靠性， 如持久化、传输确认及发布确认等。
灵活的路由 : 在消息进入队列之前，通过交换器来路由消息。对于典型的路由功能， RabbitMQ 己经提供了一些内置的交换器来实现。针对更复杂的路由功能，可以将多个 交换器绑定在一起， 也可以通过插件机制来实现自己的交换器。
扩展性: 多个RabbitMQ节点可以组成一个集群，也可以根据实际业务情况动态地扩展 集群中节点。
高可用性 : 队列可以在集群中的机器上设置镜像，使得在部分节点出现问题的情况下队 列仍然可用。
多种协议: RabbitMQ除了原生支持AMQP协议，还支持STOMP， MQTT等多种消息 中间件协议。
多语言客户端 :RabbitMQ 几乎支持所有常用语言，比如 Java、 Python、 Ruby、 PHP、 C#、 JavaScript 等。
管理界面 : RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息、集 群中的节点等。
令插件机制: RabbitMQ 提供了许多插件 ， 以实现从多方面进行扩展，当然也可以编写自 己的插件。

安装

安装erlang

 # 添加erlang solutions源
 $ wget https://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rpm
 $ sudo rpm -Uvh erlang-solutions-1.0-1.noarch.rpm
 $ sudo yum install erlang

安装rabbitMQ

$ sudo rpm --import https://www.rabbitmq.com/rabbitmq-release-signing-key.asc
$ wgethttps://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.15/rabbitmq-server-3.7.15-1.el6.noarch.rpm # 下载RabbitMQ安装包
$ sudo yum install rabbitmq-server-3.7.15-1.el6.noarch.rpm

安装依赖

sudo yum install socat

交换机

Direct exchange

将队列绑定到交换机，消息的 routeKey 需要与队列绑定的 routeKey 相同。

Fanout exchange

一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上,类似发布订阅。

Topic exchange

把消息路由到bindingKey与routingKey模糊匹配的Queue中，这种交换机是使用通配符去匹配，路由到对应的队列。通配符有两种："*" 、 "#"。需要注意的是通配符前面必须要加上"."符号。

* 符号：有且只匹配一个词。比如 a.*可以匹配到"a.b"、"a.c"，但是匹配不了"a.b.c"。
# 符号：匹配一个或多个词。比如"rabbit.#"既可以匹配到"rabbit.a.b"、"rabbit.a"，也可以匹配到"rabbit.a.b.c"。

Headers Exchange

它跟上面三种有点区别，它的路由不是用routingKey进行路由匹配，而是在匹配请求头中所带的键值进行路由。

工作模式

工作模式及交换机

• Simple Work Queue （简单工作队列），Work Queues （工作队列）
  空交换机
• Publish/Subscribe （发布订阅模式）
  fanout （扇形交换机）
• Routing（路由模式）
  direct （直连交换机）
• Topics（主题模式）
  topic（主题交换机）

简单(simple)模式

只有一个生产者、一个消费者和一个队列。生产者和消费者在发送和接收消息时，只需要指定队列名，而不需要指定发送到哪个Exchange，RabbitMQ服务器会自动使用Virtual host的默认的Exchange，默认Exchange的type为direct。

工作(work)模式

在simple模式下只有一个生产者和消费者，当生产者生产消息的速度大于消费者的消费速度时，我们可以添加一个或多个消费者来加快消费速度，这种在simple模式下增加消费者的模式，称为work模式
可以有多个消费者，但一条消息只能被一个消费者获取。
发送到队列中的消息，由服务器平均分配给不同消费者进行消费。

round robin 轮询分发

一个生产者两个消费者，消费者1 跟消费者2 处理的数据量完全一样的个数：消费者1:处理偶数消费者2:处理奇数这种方式叫轮询分发(round-robin)结果就是不管两个消费者谁忙，数据总是你一个我一个，MQ 给两个消费发数据的时候是不知道消费者性能的，默认就是雨露均沾。此时 autoAck = true。

公平分发 fair dipatch

如果要实现公平分发，要让消费者消费完毕一条数据后就告知MQ，再让MQ发数据即可。自动应答要关闭！

发布/订阅(pub/sub)模式

work模式可以将消息转到多个消费者，但每条消息只能由一个消费者获取，如果我们想一条消息可以同时给多个消费者消费呢？这时候就需要发布/订阅模式。

发布/订阅模式中，Echange的type为fanout。
生产者发送消息时，不需要指定具体的队列名，Exchange会将收到的消息转发到所绑定的队列。
消息被Exchange转到多个队列，一条消息可以被多个消费者获取。

路由(routing)模式

前面几种模式，消息的目标队列无法由生产者指定，而在路由模式下，消息的目标队列，可以由生产者指定

路由模式下Exchange的type为direct。消息的目标队列可以由生产者按照routingKey规则指定。消费者通过BindingKey绑定自己所关心的队列。一条消息队可以被多个消息者获取。只有RoutingKey与BidingKey相匹配的队列才会收到消息。

Topics 通配符模式

主题模式是在路由模式的基础上，将路由键和某模式进行匹配。其中#表示匹配多个词，星号* 表示匹配一个词，消费者可以通过某种模式的BindKey来达到订阅某个主题消息的目的.
将路由键跟某个模式匹配，# 表示匹配 >=1个字符， *表示匹配一个。生产者会带routingKey，但是消费者的MQ会带模糊routingKey。
主题模式Exchange的type取值为topic。
一条消息可以被多个消费者获取。

消息可靠性传递

消息持久化

队列和消息都支持持久化，配置不同
队列持久化需要在声明队列时添加参数 durable=True，这样在rabbitmq崩溃时也能保存队列
仅仅使用durable=True ，只能持久化队列，不能持久化消息
消息持久化需要在消息生成时，添加参数 properties=pika.BasicProperties(delivery_mode=2)

confirm机制

从producer 到 exchage 会返回一个confirmCallBakck

ACK事物机制

需要开启手动确认
消息从exchange 到 queue 投递失败会返回一个returnCallBack

死信队列

介绍

死信交换机和死信队列与普通的队列没区别，当消息成为dead message,可以被重新发送到另一个交换机这个交换机就是死信交换机。

消息成为死信的三种情况

1 队列消息长度达到限制
2 消息被拒收，并且不把消息重新放入目标队列。
3 有消息过期限制

延迟队列

介绍

延迟队列，即消息不会被立即消费，只有到达指定时间后，才会被消费。RabbitMQ没有提供延迟队列功能，可以使用过期时间+死信队列来实现延迟队列的效果。

使用场景

1 下单后，30分钟内未支付，取消订单，回滚库存。
2 新用户注册成功1天后，发送祝福短信。

消息幂等性

介绍

幂等性指一次或多次请求一个资源，对于资源本身应该具有同样的结果，也就是说，其任意多次执行对资源本身所产生的影响均与一次执行影响相同。
在MQ中指消费多条相同的消息，得到与消费该消息一次相同的结果。

解决方案

1 乐观锁解决方案

2 通过Redis原子性解决
生产者在发送消息时，为每条消息设置一个全局唯一的messageId，消费者拿到消息后，使用setnx命令，将messageId作为key放到redis中：setnx(messageId,1)，若返回1，说明之前没有消费过，正常消费；若返回0，说明这条消息之前已消费过，抛弃。
※ setnx命令，若给定的key不存在，执行set操作，返回1，若给定的Key已存在，不做任何操作，返回0。

3 唯一ID+指纹码
唯一ID+指纹码机制，利用数据库主键去重
Select count(1) from T_order where ID=唯一ID+指纹码

常用命令

查看队列： rabbitmqctl list_queues;
查看交换机： rabbitmqctl list_exchanges;
查看用户：rabbitmqctl list_users;
查看连接：rabbitmqctl list_connections;
查看消费者消息：rabbitmqctl list_consumers;
查看单个队列的内存使用：rabbitmqctl list_queues name memory;
添加用户：
--添加新用户
sudo rabbitmqctl add_user username password
--设置用户tag
sudo rabbitmqctl set_user_tags username administrator
--赋予用户默认vhost的全部操作权限
sudo rabbitmqctl set_permissions -p / username "." "." ".*"
开启web管理接口：sudo rabbitmq-plugins enable rabbitmq_management
添加开机启动RabbitMQ服务：sudo chkconfig rabbitmq-server on
启动服务:sudo /sbin/service rabbitmq-server start
查看服务状态:sudo /sbin/service rabbitmq-server status
停止服务: sudo /sbin/service rabbitmq-server stop
查看默认配置：rabbitmqctl status

[消息队列] RabbitMQ

标签：server   流量   工作   mq服务器   format   请求   return   区别   异步处理   

原文地址：https://www.cnblogs.com/gustavo/p/304e50a74a15c2767395c4a9cee119b9.html