spring-rabbit消费过程解析及AcknowledgeMode选择

古城微笑少年丶 2024-05-02 20:14 49阅读 0赞

说明：本文内容来源于对amqp-client和spring-rabbit包源码的解读及debug，尽可能保证内容的准确性。

rabbitmq消费过程示意如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RydXNhdXNl_size_16_color_FFFFFF_t_70][]

图中首字母大写的看上去像类名的，如ConsumerWorkService，MainLoop，WorkPoolRunnable等，没错就是类名，可自行根据类名去查看相关源码。

下面解释上图的含义。

1. 启动流程  
通过BeanPostProcessor扫描所有的bean中存在的@RabbitListener注解及相应的Method；  
由RabbitListenerContainerFactory根据配置为每一个@RabbitListener注解创建一个MessageListenerContainer，持有@RabbitListener注解及Method信息；  
初始化MessageListenerContainer，主要是循环依次创建consumer（AsyncMessageProcessingConsumer类），启动consumer；  
创建consumer，过程包括：创建AMQConnection（仅第一次创建），创建AMQChannel（每个consumer都会创建），发送消费queue的请求（basic.consume），接收并处理消息；  
AMQConnection持有连接到rabbitmq server的Socket，创建完成后启动MainLoop循环从Socket流中读取Frame，此时流中没有消息，因为channel还没创建完成；  
创建AMQChannel（一个AMQConnection中持有多个AMQChannel），并将创建完成的channel注册到AMQConnection持有的ConsumerWorkService，实际就是添加到WorkPool类的Map里面去，此时Socket流中也没有消息，因为channel还没有与queue绑定；  
创建完成的AMQChannel的代理返回给consumer，consumer通过channel发送消费queue的请求到rabbitmq server（绑定成功），此时还没开始处理消息，但Socket流中已经有消息，并且已经被connection读取到内存（即BlockingQueue<Runnable>）中，并且已经开始向BlockingQueue<Delivery>分发；  
consumer启动循环，从BlockingQueue<Delivery>中取消息，利用MessageListenerContainer中持有的Method反射调用@RabbitListener注解方法处理消息。  
2. 消费流程  
rabbitmq server往Socket流中写入字节。  
AMQConnection启动一个main loop thread来跑MainLoop，不断从Socket流中读取字节转换成Frame对象，这是每个connection唯一的数据来源。  
Frame对象结构如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RydXNhdXNl_size_16_color_FFFFFF_t_70 1][]

type：指定当前Frame的类型，如method(1)、message header(2)、message body(3)、heartbeat(8)等；

channel：channel的编号，从0~n排列，指定当前Frame需要交给哪个channel处理。channel-0为一类，channel-n为一类。channel-0是一个匿名类，用来处理特殊Frame，如connection.start。channel-n都是ChannelN类，由ChannelManager类统一管理。

payload：当前Frame的具体内容。

consumer启动后，connection读取到的Frame如上图所示（一个consumer的情况，多个consumer的Frame可能会交替）。从basic.deliver开始是消息的内容，每条消息分成三个Frame：第一个是method，basic.deliver代表这是一个消息，后面一定会再跟着两个Frame；第二个是message header；第三个是message body，body读取之后将三个Frame整合到一起转换成一条完整的deliver命令。

AMQConnection根据读取到的Frame中的type决定要怎么处理这个Frame：heartbeat(8) do nothing；其它的根据channel编号交给相应的AMQChannel去处理，（编号为0的是特殊的channel，消息相关的用的都是编号非0的channel），消息都会拿着这个编号到ChannelManager找对应的ChannelN处理。  
ChannelN经过一系列中间过程由Frame（消息是三个Frame）得到了Runnable，将(ChannelN, Runnable) put到ConsumerWorkService持有的WorkPool里面的一个Map<Channel, BlockingQueue<Runnable>>里面去。这样这个Runnable就进入了与ChannelN对应的BlockingQueue<Runnable>（写死的size=1000）里面了。  
execute一个WorkPoolRunnable，执行的任务是：从WorkPool中找出一个ready状态的ChannelN，把这个ChannelN设为inProgress状态，从对应的BlockingQueue<Runnable>中取最多16（写死的）个Runnable在WorkPoolRunnable的线程里依次执行（注意：此处不再另开线程，所以可能会堵塞当前线程，导致这个ChannelN长时间处于inProgress状态），执行完后将当前ChannelN状态改为ready，并在当前线程execute另一个WorkPoolRunnable。  
BlockingQueue<Runnable>里面的Runnable执行的逻辑是：构造一个Delivery put到与ChannelN对应的AsyncMessageProcessingConsumer持有的BlockingQueue<Delivery>（size=prefetchCount可配置）里面去（如果消息处理速度太慢，BlockingQueue<Delivery>已满，此处会堵塞）。  
每个AsyncMessageProcessingConsumer都有一个独立的线程在循环从BlockingQueue<Delivery>一次读取一个Delivery转换成Message反射调用@RabbitListener注解方法来处理。  
3. 无ack消费模式与有ack消费模式对比  
根据以上对消费过程的分析，将无ack模式与ack模式进行对比。

无ack模式（AcknowledgeMode.NONE）  
server端行为  
rabbitmq server默认推送的所有消息都已经消费成功，会不断地向消费端推送消息。  
因为rabbitmq server认为推送的消息已被成功消费，所以推送出去的消息不会暂存在server端。  
消息丢失的风险  
当BlockingQueue<Runnable>堆满时（BlockingQueue<Delivery>一定会先满），server端推送消息会失败，然后断开connection。消费端从Socket读取Frame将会抛出SocketException，触发异常处理，shutdown掉connection和所有的channel，channel shutdown后WorkPool中的channel信息（包括channel inProgress,channel ready以及Map）全部清空，所以BlockingQueue<Runnable>中的数据会全部丢失。

此外，服务重启时也需对内存中未处理完的消息做必要的处理，以免丢失。

而在rabbitmq server，connection断掉后就没有消费者去消费这个queue，因此在server端会看到消息堆积的现象。

有ack模式（AcknowledgeMode.AUTO，AcknowledgeMode.MANUAL）  
AcknowledgeMode.MANUAL模式需要人为地获取到channel之后调用方法向server发送ack（或消费失败时的nack）信息。

AcknowledgeMode.AUTO模式下，由spring-rabbit依据消息处理逻辑是否抛出异常自动发送ack（无异常）或nack（异常）到server端。

server端行为  
rabbitmq server推送给每个channel的消息数量有限制，会保证每个channel没有收到ack的消息数量不会超过prefetchCount。  
server端会暂存没有收到ack的消息，等消费端ack后才会丢掉；如果收到消费端的nack（消费失败的标识）或connection断开没收到反馈，会将消息放回到原队列头部。  
这种模式不会丢消息，但效率较低，因为server端需要等收到消费端的答复之后才会继续推送消息，当然，推送消息和等待答复是异步的，可适当增大prefetchCount提高效率。

注意，有ack的模式下，需要考虑setDefaultRequeueRejected(false)，否则当消费消息抛出异常没有catch住时，这条消息会被rabbitmq放回到queue头部，再被推送过来，然后再抛异常再放回…死循环了。设置false的作用是抛异常时不放回，而是直接丢弃，所以可能需要对这条消息做处理，以免丢失。更详细的配置参考这里。

对比  
无ack模式：效率高，存在丢失大量消息的风险。  
有ack模式：效率低，不会丢消息。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RydXNhdXNl_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/02/d7633b9ba6034f449d8833dba788f0d0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RydXNhdXNl_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/02/68535a683e63424c9d4fef5aa0e2c4b7.png