RabbitMQ实战-高效部署分布式消息队列

一、天降奇兵

1.消息队列使用消息将应用程序连接起来。这些消息通过像RabbitMQ这样的消息代理服务器在应用程序之间路由

2.RabbitMQ优势：

• 除Qpid外，唯一实现了AMQP标准的代理服务器
• 由于Erlang，RabbitMQ集群不可思议的简单
• 比其他更可靠，更能防止崩溃

3.检查服务器状态：rabbitmqctl status

二、理解消息通信

A.消费者和生产者

1.RabbitMQ不是快餐车而是消息投递服务，在应用程序和服务器之间扮演着路由器的角色

2.生产者（producer）创建消息，然后发布（发送）到代理服务器（RabbitMQ）。消息包含有效载荷（payload）和标签（label）

• 有效载荷就是你要传输的数据，可以是任何内容
• 标签描述了有效载荷，并用它来决定谁将获得消息的拷贝

3.消息者，连接到代理服务器上，并订阅到队列（queue）上，当消费者接收到消息时，它只得到消息的一部分：有效载荷

4.生产者创建消息，消费者接收这些消息。应用程序可以作为生产者，向其他应用程序发送消息。或者作为一个消费者，接收消息。也可以在两者之间进行切换。不过必须先建立一条信道（channel）

5.在应用程序和Rabbit代理服务器之间创建一条TCP连接。一旦TCP连接找开（通过了认证），应用程序就可以创建一条AMQP信道。信道是建立在“真实的”TCP连接内的虚拟连接，每条信道都会被指派一个唯一ID

B.从底部开始构造：队列

1.AMQP消息路由必须有三部分：交换器、队列和绑定

• 生产者把消息发布到交换器上；消息最终到达队列，并被消费者接收；绑定决定了消息如何从路由器路由到特定的队列

2.队列就如同具名邮箱，消息最终达到队列中并等待消费，消费者通过以下两种方式从特定队列中接收消息：

• 通过AMQP的basic.consume命令订阅：如果消费者处理队列消息，并且/或者需要在消息一到达队列时就自动接收的话，应该使用这个命令
• basic.get命令：会订阅消息，获得单条消息，然后取消订阅，不要放在一个循环中来代替basic.consume，会影响性能

3.当Rabbit队列拥有多个消费者时，队列收到的消息将以循环（round-robin）的方式发送给消费者，每条消息只会发送给一个订阅的消费者

4.消费者必须通过AMQP的basic.ack显示地向RabbitMQ发送一个确认，或者在订阅到队列的时候就将auto_ack参数设置为true。消费者对消息的确认和告诉生产者消息已经被接收了这两件事毫不相关。消费者通过确认命令告诉RabbitMQ它已经正确地接收了消息，同时RabbitMQ才能安全地把消息从队列中删除

5.如果消费者收到一条消息，然后确认之前从Rabbit断开连接（或者从队列上取消订阅），RabbitMQ会认为这条消息没有分发，然后重新分发给下一个订阅的消费者

6.拒绝消息：

• 把消费者从RabbitMQ服务器断开连接：会导致RabbitMQ自动重新把消息入队并发送给另一个消费者，缺点是连接/断开连接的方式会额外增加RabbitMQ的负担
• RabbitMQ2.0.0以上版本可以使用AMQP的basic.reject命令

7.消费者和生产者都能使用AMQP的queue.declare命令来创建队列

• 如果消费者在同一条信道上订阅了一另一个队列的话，就无法再声明队列了，必须先取消订阅并将信道置为“传输”模式
• 消费者订阅队列时需要队列名称，并在创建绑定时也需要指定队列名称
• exclusive：如果设置为true，队列变成私有的，只有自己的应用程序才能消费队列
• auto-delete：当最后一个消费者取消订阅的时候，队列就会自动移除
• passive：设置为true后，如果队列存在，queue.declare就会成功返回；如果不存在，会返回错误

8.如果不能承担得起消息进入“黑洞”而丢失的话，生产者和消费者都应该尝试去创建队列；否则可以只让消费者来声明队列

C.联合起来：交换器和绑定

1.当你把消息发送到代理服务器时，消息将拥有一个路由键——即使是空的——RabbitMQ也会将其和绑定使用的路由键进行匹配。如果相匹配的话，那么消息将会投递到该队列。如果不匹配任何绑定模式的话，消息将进入“黑洞”

2.交换器和绑定可以完成不同的使用场景；对于发送消息给服务器的发布者，不需要关心服务器另一端（整个消息处理环节中的队列和消费者）的逻辑

3.处理投递到多个队列的协议：

• direct交换器：如果路由键匹配的话，消息就投递到对应的队列。服务器必须实现direct类型交换器，包含一个空白字符串名称的默认交换器。当声明一个队列时，它会自动 绑定到默认交换器，并以聊表名称作为路由键 $channel->basic_publish(消息内容,默认交换器,路由键)
• fanout交换器：会将收到的消息广播到绑定的队列上，当发送一条消息到fanout交换器时，会把消息投递给所有附加在此交换器上的队列
• topic交换器：使得来自不同源头的消息能够到达同一个队列

D.多租户模式：虚拟主机和隔离

1.每一个RabbitMQ服务器都能创建虚拟消息服务器，称为虚拟机（vhost），每一个vhost本质上是一个mini版的RabbitMQ服务器，拥有自己的队列、交换器和绑定，以及权限机制

2.在Rabbit里创建一个用户时，用户通常会被指派给至少一个vhost，并且只能访问被指派vhost内的队列、交换器和绑定，在设计消息通信架构时，记住vhost之间是绝对隔离的，在集群上创建vhost时，整个集群都会创建该vhost

3.创建、删除、查看：

• rabbitmqctl add_vhost [vhost_name]
• rabbitmqctl delete_vhost [vhost_name]
• rabbitmqctl list_vhosts

E.持久化和策略

1.队列和交换器的durable属性设为true，决定RabbitMQ在崩溃或重启之后重新创建队列（或交换器）

2.消息想要从Rabbit崩溃中恢复，必须：

• 把它的投递模式（delivery mode）选项设置为2（持久）
• 发送到持久化的交换器
• 到达持久化的队列

3.确保持久性消息能恢复的方式是，将它们写入磁盘上的一个持久化日志文件，当发布一条持久性消息到持久交换器上时，Rabbit会在消息提交到日志文件后才发送响应，之后这条消息如果路由到了非持久队列的话，它会自动从持久性日志中移除

4.持久性的劣势：性能变慢，内建集群中工作得不好

5.对于关键消息使用持久化机制

6.AMQP事务会大大降低Rabbit的性能，Rabbit团队使用发送方确认模式，将信道设置成confirm模式，是异步的

https://github.com/zhangyue0503/rabbitmq/tree/master/2

三、运行和管理Rabbit

A.服务器管理

1.节点：描述的是一个Erlang节点运行着一个Erlang应用程序，多个应用程序可以运行在同一个节点之上，RabbitMQ节点指的是RabbitMQ应用程序和其所在的Erlang节点

2.rabbitmq-server 启动；rabbitmqctl stop 关闭；

3.配置文件rabbitmq-config，vm_memory_high_watermark指定可消耗的内存，十进制数字，0.4表示为40%

B.请求许可

1.用户相关命令：

• rabbitmqctl add_user 用户名 密码
• rabbitmqctl delete_user 用户名
• rabbitmqctl list_users
• rabbitmqctl change_password 用户名 新密码

2.访问控制条目组成：被授予访问权限的用户、权限控制应用的vhost、需要授予的读/写/配置权限的组合、权限范围

3.访问控制条目是无法跨越vhost的

4.访问权限命令：

• rabbitmqctl set_permissions -p [vhost] [user] "配置" "写" "读"，".*"匹配任何队列和交换器，"check-.*"匹配以check-开头的队列和交换器，""不匹配队列和交换器
• rabbitmqctl list_permissions -p [vhost] 验证权限是否正确
• rabbitmqctl clear_permissions -p [vhost] [user]移除用户在vhost上的权限
• rabbitmqctl list_user_permissions [user]查看用户在所有vhost上的权限

C.检查

1.相关命令：

• rabbitmqctl list_queues -p [vhost] [name名称 messages消息数目 consumers消费者数目 memory内存使用等……]列出所有队列及队列里的消息数目
• rabbitmqctl list_exchanges 查看交换器信息
• rabbitmqctl list_bindings 查看绑定信息

2.日志

• *-sasl.log，SASL（系统应用程序支持库）是库的集合，帮助开发Erlang应用时，提供一系列标准
• 轮换日志：rabbitmqctl rotate_logs shuffix，shuffix通常是数字，添加到被轮换日志文件的末尾

D.修复Rabbit：疑难解答

1.RabbitMQ使用Mnesia数据库存储队列、交换器、绑定等信息

https://github.com/zhangyue0503/rabbitmq/tree/master/3

四、解决Rabbit相关问题：编码与模式

A.解耦

1.异步状态思维（分离请求和动作）：接收请求->RabbitMQ->处理请求（入库）

2.自动轮询（round-robin）可代替负载均衡

3.零成本API：语言不应成为枷锁

4.如何将应用切分？应用程序的哪部分是订单接收者，哪部分是订单处理者？

B.发后即忘模型

1.创建了任务，放置到交换器上，并让你的应用程序返回继续工作

2.匹配该模式的两种一般类型的任务：

批处理（batch processing）：针对大型数据集合的工作或者转换

通知（notifications）：以发生事件的描述

3.RabbitMQ使用“.”作为标记中不同部分的分隔符；no_ack=false告诉RabbitMQ要显示确认收到的消息，这将暂停从队列发送新的消息过来，直到收到的最后一条消息处理完成并发送确认消息为止；

C.用RabbitMQ实现RPC并等待响应

1.使用reply_to作为发布应答消息的目的地，同时发布的时候无须指定交换器

2.exclusive=true，确保只有自己才能读取队列上的数据；auto_delete，接收完消息后断开队列的连接时，Rabbit会自动将队列删除

https://github.com/zhangyue0503/rabbitmq/tree/master/4

五、集群并处理失败

A.集群架构

1.集群只会在单个节点上而不是所有节点上创建完整的队列信息（元数据、状态、内容）

2.交换器只是一个名称和一个队列的绑定列表，信道才是真正的路由器

3.使用AMQP事务，在消息路由到队列之前会一直阻塞；或者使用发送方确认（publisher confirm）模式来记录连接中断时尚未被确认的消息。这两种解决方案可以帮助在节点故障并且目的队列不复存在时检测到消息无法路由的情况

4.单节点必须是磁盘类型节点，否则一重启所有配置信息都会丢失；集群允许只有一个节点是磁盘节点，其他可以是内存节点，当磁盘节点崩溃后，不能对队列进行操作；可以设置两个磁盘节点；添加内在节点时，需要确保告知所有的磁盘节点；

B.单机集群

1.需要修改环境变量RABBITMQ_NODE_PORT和RABBITMQ_NODENAME，如RABBITMQ_NODE_PORT=5763和RABBITMQ_NODENAME=rabbit_1，然后rabbitmq-server -detached

2.rabbitmqctl -n rabbit_1@主机名 stop_app，然后再reset

3.rabbitmqctl -n rabbit_1@主机名 join_cluster rabbit@主机名

4.rabbitmqctl -n rabbit_1@主机名 start_app

5.rabbitmqctl cluster_status

6.-n表示指定节点而非默认节点上执行命令

C.将节点分布到更多的机器上

1.需要复制找到.erlang.cookie，复制其中的字符串到其他节点上，然后再进行join_cluster

2.删除节点，直接通过reset命令

D.镜像队列和保留消息

1.镜像队列的主拷贝仅存在于一个节点（主队列，master），在其他的集群上拥有从队列（slave）拷贝

2.xa-ha-policy和xa-ha-policy-params

https://github.com/zhangyue0503/rabbitmq/tree/master/5

六、从故障中恢复

A.为Rabbit做负载均衡

1.当为Rabbit添加负载均衡器时，集群节点就作为负载均衡器背后的服务器，而你的生产者和消费者就是客户，应用程序只需知道负载均衡器的前端IP；负载均衡器会以最小的连接负载透明地将客户端连接到集群节点

B.连接丢失和故障转移

1.应该总是将故障转移视为连接到了一个完全 无关的RabbitMQ服务器，而不是有着共享状态的集群节点，不论节点故障什么时候发生，在检测到故障并进行重边之后的首要任务是构造交换器、队列和绑定

七、warren和Shovel：故障转移和复制

A.warren：另一种集群方式

1.一个warren是指一对主/备独立服务器，并前置一台负载均衡器来处理故障转移。这是真正的无共享架构，主和备之间没有协作。

B.远距离通信和复制

1.Shovel是RabbitMQ的一个插件，能够定义RabbitMQ上的队列和另一个RabbitMQ上的交换器之间的关系

2.安装插件：rabbitmq-plugins enable rabbitmq_shovel

八、从Web端管理RabbitMQ

A.超越rabbitmqctl：RabbitMQ Management插件

1.rabbitmq-plugins enable rabbitmq_management，http://localhost:15672/mgmt/

2.rabbitmqadmin脚本：wget http://localhost:15672/cli/rabbitmqadmin

九、使用REST API控制Rabbit

1.http://localhost:15672/api，如http://localhost:15672/api/users

十、监控

1.四种整型退出代码：0-OK，1-WARNING，2-CRITICAL，3-UNKNOWN

十一、提升性能，保障安全

1.如果可以接受丢失消息，将delivery-mode设置为1，设置为2将持久化，消息写写到磁盘上，降低速度

2.no-ack为true，服务器会在消息发送给客户端后自动将其出队

3.direct交换器和fanout交换器的差别在于后者在查询rabbit_route表时忽略了路由键；topic交换器相比前两者会占用更多的内存

4.消息中的mandatory和immediate标记为false的话，会以异步方式投递消息

5.RabbitMQ被优化为尽可能快地向消费者投递消息，应该尽可能让队列保持为空

6.如果队列声明中durable为true的话，会在表rabbit_queue和rabbit_durable_queue添加队列记录，否则只会在rabbit_queue中存放记录

十二、扩展RabbitMQ

1.http://www.rabbitmq.com/plugins.html

2.插件启用：rabbitmq-plugins enable xxxxx；移除插件：rabbitmq-plugins disable xxx