RabbitMQ生产故障问题分析

1. 问题引发

　　由某个服务BI-collector-xx队列出现阻塞，影响很整个rabbitMQ集群服务不可用，多个应用MQ生产者服务出现假死状态，系统影响面较广，业务影响很大。当时为了应急处理，恢复系统可用，运维相对粗暴的把一堆阻塞队列信息清空，然后重启整个集群。

在复盘整个故障过程中，我心中有不少疑惑，至少存在以下几个问题点：

1. 为什么出现队列阻塞？
2. 某个队列出现阻塞为什么会影响到其他队列的运行（即多队列间相互影响）？
3. 某个应用MQ队列出现问题，为什么会导致应用不可用呢？

2. 试验队列阻塞

某天周末在家里，找个测试环境，安装rabbitmq尝试重现这过程，并做模拟测试。

写两个测试应用Demo（假设是两个项目应用）分别有生产者和消费者，并分别使用队列testA和testB。

为了尽可能还原生产的情况，一开始测试使用了同一个vhost，后面分别设置不同vhost。

生产者A，示例代码如下

消费者A

MQ配置

生产者B，每次生产10万条消息

 

消费者B，代码故意写错（模拟出现异常的情况），不是正常的json串导致解释json时抛出异常

先了解一下Rabbitmq客户端启动连接工作过程，通过wireshark抓包分析，如下

 

先对AMQP做一个简单的介绍，请求的AMQP协议方法信息，AMQP协议方法包含类名+方法名+参数，这一列主要展示了类名和方法名

• Connection.Start：请求服务端开始建立连接
• Channel.Open：请求服务端建立信道
• Queue.Declare：声明队列
• Basic.Consume：开始一个消费者，请求指定队列的消息

详细方法可以查看amqp官网https://www.rabbitmq.com/amqp-0-9-1-reference.html

工作过程分析：

Basic.Publish： 客户端发送Basic.Publish方法请求，将消息发布到exchange，rabbitmq server会根据路由规则转发到队列中；

Basic.Deliver： 服务端发送Basic.Deliver方法请求，投递消息到监听队列的客户端消费者；

Basic.Ack： 客户端发送Basic.Ack方法请求，告知rabbimq server,消息已接收处理。

两个应用程序启动后，通过rabbitmq管理控制台可以观察一些参数和监控指标

 

 

一开始A应用生产和消费都是正常的。

B消费端错误代码异常，狂刷报错信息

 

经过大概30分钟运行，观察A生产者应用控制台也有出现异常信息

 

查看服务端连接状态出现blocked情况，与生产故障发生情景很类似。

 

此时客户端即本机器，CPU和内存上涨明显，风扇声音很响，明显卡顿，再过30分钟应用基本不可用状态。

分析原因

上面错误代码展示了消费者B无法ack，由于没有进行ack导致队里阻塞。那么问题来了，这是为什么呢？其实这是RabbitMQ的一种保护机制。防止当消息激增的时候，海量的消息进入consumer而引发consumer宕机。

 RabbitMQ提供了一种QOS(服务质量保证)功能，即在非自动确认的消息的前提下，限制信道上的消费者所能保持的最大未确认的数量。可以通过设置prefetchCount实现，自动确认prefetchCount设置无效。

举例说明:可以理解为在consumer前面加了一个缓冲容器，容器能容纳最大的消息数量就是PrefetchCount。如果容器没有满RabbitMQ就会将消息投递到容器内，如果满了就不投递了。当consumer对消息进行ack以后就会将此消息移除，从而放入新的消息。

通过上面的配置发现prefetch初始我只配置了2，并且concurrency配置的只有1，所以当我发送了2条错误消息以后，由于解析失败这2条消息一直没有被ack。将缓冲区沾满了，这个时候RabbitMQ认为这个consumer已经没有消费能力了就不继续给它推送消息了，所以就造成了队列阻塞。

判断队列是否有阻塞的风险。

  当ack模式为manual，并且线上出现了unacked消息，这个时候不用慌。由于QOS是限制信道channel上的消费者所能保持的最大未确认的数量。所以允许出现unacked的数量可以通过channelCount * prefetchCount *消费节点数量得出。

channlCount就是由concurrency,max-concurrency决定的。

• min = concurrency * prefetch *消费节点数量
• max = max-concurrency * prefetch *消费节点数量

由此可以得出结论

• unacked_msg_count < min 队列不会阻塞。但需要及时处理unacked的消息。
• unacked_msg_count >= min 可能会出现堵塞。
• unacked_msg_count >= max 队列一定阻塞。

重点注意

1、unacked的消息在consumer切断连接后(如重启)再连接，会自动回到队头。

2、若将ack模式改成auto自动，这样会使QOS不生效。会出现大量消息涌入consumer从而可能造成consumer宕机风险。

再回看程序配置，做一些分析和调整

对B消费端问题代码加个try-catch-finally，不管中间有何问题，都进行消息签收ACK。

 

代码调整之后，两个队列正常运行，客户端两个应用也正常运行。

 

 

经过一段时间消费，B消费者端已经把堆积的消息消费完了。

 

3、    第三个问题原因分析

还是查看抓包信息

Basic.Reject： 客户端发送Basic.Reject方法请求，表示无法处理消息，拒绝消息，此时的requeue参数为true，将消息返回原来的队列；

Basic.Deliver： 服务端调用Basic.Deliver方法，和第一次Basic.Deliver方法不同的是，此时的redeliver参数为true，表示重新投递消息到监听队列的消费者，然后这两步会一直重复下去。

RabbitMQ消息监听程序异常时，consumer会向rabbitmq server发送Basic.Reject，表示消息拒绝接受，由于Spring默认requeue-rejected配置为true，消息会重新入队，然后rabbitmq server重新投递。就相当于死循环了，所以容易导致消费端资源占用过高，特别是TCP连接数、线程数、IO飙升，如果个别程序带事务或数据库操作等连接资源得不到释放也会占满，导致应用假死状态（出现问题的时候，查看问题应用出现大量的connection timeout错误报错日志）。

因此针对性的，有些业务场景（不强调数据强一致性的场景，比如日志收集）可以设置default-requeue-rejected: false即可。

factory.setDefaultRequeueRejected(false);

  会根据异常类型选择直接丢弃或加入dead-letter-exchange中。

消费者端正确的使用手动确认示例结构代码，很重要！

try {// 业务逻辑。
}catch (Exception e){// 输出错误日志。
}finally {// 消息签收。
}

4、    验证队列设置最大长度限制

设置queueLengthLimit队列最大长度限制 x-max-length=5

 

生产者原本想要生产10条消息

 

由于受到队列最大长度限制，实际上只有5条入队列里面。

 

消费者拿出来的消息，仅有5条，从NO.6~NO.10

改变消费者程序，让生产者一直产生消息，消费者消费速度明显赶不上生产者的生产速度。

 

 

从消费端来看消息是随机性入队的，队列里面一直最多5条消息，发再多也进不了，消息者和生产者也不会发生什么异常，只是消息会随机性丢失（并没有全部入队）。

运行情况良好，除了消息没有全部入队列 ，没有出现异常情况

 

消费比较慢，本机器CPU和内存各项指标正常，没有异常。

搞一个异常情况出现unack，最大队列长度限制，是不算unack数量的，如下图所示

 

异常之后，此观察MQ监控管理后台

 

生产者不停一直在生产消息，运行30分钟，观察生产者应用也是正常的的，就是消息入不了队列。

 

 

5、  检查实际的业务端代码

再看我们业务系统消费端代码，消费端各种不规范写法都有，以下例举几个典型

1、手动签收有ACK，但是没有try-catch-finally结构，消费端业务代码如下：

2、有try-catch-finally结构，但是deliverTag是一个固定值0，一样的会出问题。

 

3、自动签收确认的，大量消息的时候，容易搞死消费端应用。

 

6. 总结

• 生产环境不建议使用自动ack模式，这样会使QOS无法生效。
• 在使用手动ack的时候，需要非常注意消息签收，业务代码使用try-catch-finally处理结构，防止业务代码异常时无法签收。
• 规范约束mq客户端代码，正确的使用Rabbitmq配置。
• 不同业务项目设置不同的vhost可以隔离一些影响，提升rabbitmq资源使用。
• 考虑设置dead-letter-exchange，当设置了 requeue=false时，可以放入dead-letter-exchange，可以快速排查定位问题。
• Exchange和队列的最大长度限制可以是限制消息的数量（参数：x-max-length），或者是消息的总字节数（总字节数表示的是所有的消息体的字节数，忽略消息的属性和任何头部信息），又或者两者都进行了限制，两者取小值生效，只有处于ready状态的消息被计数，未被确认的消息不会被计数受到limit的限制。最大队列设置可以限制生产端，但会造成消息丢失风险，最大消息数量限制，不能完全解决队列阻塞问题。
• 尽量使用Direct-exchange，Direct 类型的 Exchange 投递消息是最快的。
  • Direct：处理路由键，需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键为“A”，则只有路由键为“A”的消息才被转发，不会转发路由键为"B"，只会转发路由键为“A”；
  • Topic：将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“*”只能匹配一个词；
  • Fanout：不处理路由键。只需要简单的将队列绑定到交换机上。一个发送到该类型交换机的消息都会被广播到与该交换机绑定的所有队列上；
  • Headers：不处理路由键，而是根据发送的消息内容中的 headers 属性进行匹配。在绑定 Queue 与 Exchange 时指定一组键值对；当消息发送到 RabbitMQ 时会取到该消息的 headers 与 Exchange 绑定时指定的键值对进行匹配；如果完全匹配则消息会路由到该队列，否则不会路由到该队列。