消息队列RabbitMQ和ActiveMQ的生产者流量控制

现在的位置: 首页 > 综合 > 正文

RSS

消息队列RabbitMQ和ActiveMQ的生产者流量控制

2013年03月30日 ⁄ 综合 ⁄ 共 2686字 ⁄ 字号小中大 ⁄ 评论关闭

20120825 郑昀

Q：MQ 们为什么要做生产者流量控制？

A：麻烦就在于：『像 Erlang 的虚拟机实现和设计上都没有阻止用户往一个进程的消息队列里扔消息，当消息的生产速度过快，超过进程的处理能力时，这些消息就堆积起来，占用越来愈多的内存，最终导致VM崩溃。』

Q：我为什么要知道 MQ 在做生产者流量控制？

A：当你发现自家的 Producers 动辄被挂起或被阻塞时，你要知道该调 Consumer 的消费速率，还是调 Memory Threshold of MQ 。

一，RabbitMQ 2.8.0+的流量控制

RabbitMQ 2.8.0+引入了一个新特性“internal flow control”。

至此 RabbitMQ 有三种流量控制：

1.1. Per-Connection Flow Control

是面向每一个连接做的流量控制。

即，RabbitMQ 会主动阻塞（Block）那些发布消息太快的连接（Connections），无需做任何配置。

如果连接被阻塞了，那么它在 rabbitmqctl 控制台上会显示一个blocked的状态。

RabbitMQ 的流量控制机制是基于信用证(Credit)的拥塞控制机制，后面郑昀会在附录A中列出这个古老控制机制的详细介绍。

1.2. Memory-Based Flow Control

RabbitMQ 会在启动时检测机器的物理内存数值。默认当 MQ 占用 40% 以上内存时，MQ 会主动抛出一个内存警告并阻塞所有连接（Connections）。

你也可以通过修改 rabbitmq.config 文件来调整内存阈值，默认值是 0.4，如下所示：

[{rabbit, [{vm_memory_high_watermark, 0.4}]}].

当 MQ 启动时，该内存阈值也会写入到 RABBITMQ_NODENAME.log 文件里，如下所示：

Memory limit set to 2048MB.

如果 MQ Server 不能识别你的系统，或者你在用 Windows 系统，那么它会写一个警告信息到 RABBITMQ_NODENAME.log 文件里，如下所示：

=WARNING REPORT==== 29-Oct-2009::17:23:44 ===
Unknown total memory size for your OS {unix,magic_homebrew_os}. Assuming memory size is 1024MB.

1.3. Disk-Based Flow Control

默认情况，如果剩余磁盘空间在 1GB 以下，RabbitMQ 主动阻塞所有的生产者。这个阈值也是可调的。

二，Apache ActiveMQ 的流量控制

2.1. ActiveMQ 的生产者流量控制的触发条件有三个：

不管 mq 有无做持久化配置：

！ActiveMQ所使用的内存到达 memoryUsage 配置值，默认值64MB；

如果 mq 做了持久化配置：

！要打开了useCache开关，表明要将持久化消息缓存起来以便快速访问，默认是True；
！缓存在内存中消息总字节数到达 memoryLimit 配置值，默认值是1MB。

2.2. 触发之后会导致：

生产者调用发送消息函数时被Stuck（卡住）。

（注：关键词 activemq+stuck+producer 或 activemq+block+producer）

附录A

一，基于信用证(Credit)的拥塞控制机制

1993年，ATM领域开始寻找一种可以动态分配带宽并防止信元丢失的传输机制。因此提出了ABR（Available Bit Rate，可用比特率）业务，ABR 不强制网络为其分配固定的带宽，网络通过反馈信息动态调整分配给每个 ABR 连接的带宽。很多专家都投入到 ABR 业务的拥塞控制规范研究上。

基于 Credit 的方案由哈佛大学的H．T．Kung教授首先提出，采用链路级流控机制，以单一链路或虚电路VC作为基本的控制单位。每条链路有一个信元发送节点（可以是一个源端或一个交换节点）和一个接收节点（可以是一个交换节点或一个目的端系统），每个节点为每一条VC维持一个排队队列，信元接收端监视每条VC的排队队列长度，决定发送端可以发到VC上的信元数量（用信用证通知），信元发送端只能发送信用证值所允许的最大信元数量，如果只有一条VC，则信用证的值应足够大，以充分利用链路的带宽。一般来说，信用证值应满足下列条件：信用证值大于等于链路信元速率乘以链路来回程传输时延。