RabbitMQ消息的传输控制

た入场券 2022-07-14 01:59 141阅读 0赞

# 1、消息机制 #

## 1.1消息的应答 ##

RabbitMQ有两种应答模式，自动和手动。这也是AMQP协议所推荐的。这在point-to-point和broadcast都是一样的。

自动应答-当RabbitMQ把消息发送到接收端，接收端把消息出队列的时候就自动帮你发应答消息给服务。

手动应答-需要我们开发人员手动去调用ack方法去告诉服务已经收到。

文档推荐在大数据传输中，如果对个别消息的丢失不是很敏感的话选用自动应答比较理想，而对于那些一个消息都不能丢的场景，需要选用手动应答，也就是说在正确处理完以后才应答。如果选择了自动应答，那么消息重发这个功能就没有了。

## 1.2消息的拒收 ##

拒收，是接收端在收到消息的时候响应给RabbitMQ服务的一种命令，告诉服务器不应该由我处理，或者拒绝处理，扔掉。接收端在发送reject命令的时候可以选择是否要重新放回queue中。如果没有其他接收者监控这个queue的话，要注意一直无限循环发送的危险。

BasicDeliverEventArgs ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

channel.BasicReject(ea.DeliveryTag, false);

BasicReject方法第一个参数是消息的DeliveryTag，对于每个Channel来说，每个消息都会有一个DeliveryTag，一般用接收消息的顺序来表示：1,2,3,4 等等。第二个参数是是否放回queue中，requeue。

BasicReject一次只能拒绝接收一个消息，而BasicNack方法可以支持一次0个或多个消息的拒收，并且也可以设置是否requeue。

channel.BasicNack(3, true, false);

在第一个参数DeliveryTag中如果输入3，则消息DeliveryTag小于等于3的，这个Channel的，都会被拒收。

## 1.3消息的QoS ##

QoS = quality-of-service， 顾名思义，服务的质量。通常我们设计系统的时候不能完全排除故障或保证说没有故障，而应该设计有完善的异常处理机制。在出现错误的时候知道在哪里出现什么样子的错误，原因是什么，怎么去恢复或者处理才是真正应该去做的。在接收消息出现故障的时候我们可以通过RabbitMQ重发机制来处理。重发就有重发次数的限制，有些时候你不可能不限次数的重发，这取决于消息的大小，重要程度和处理方式。

甚至QoS是在接收端设置的。发送端没有任何变化，接收端的代码也比较简单，只需要加如下代码：

channel.BasicQos(0, 1, false);

代码第一个参数是可接收消息的大小的，但是似乎在客户端2.8.6版本中它必须为0，即使：不受限制。如果不输0，程序会在运行到这一行的时候报错，说还没有实现不为0的情况。第二个参数是处理消息最大的数量。举个例子，如果输入1，那如果接收一个消息，但是没有应答，则客户端不会收到下一个消息，消息只会在队列中阻塞。如果输入3，那么可以最多有3个消息不应答，如果到达了3个，则发送端发给这个接收方得消息只会在队列中，而接收方不会有接收到消息的事件产生。总结说，就是在下一次发送应答消息前，客户端可以收到的消息最大数量。第三个参数则设置了是不是针对整个Connection的，因为一个Connection可以有多个Channel，如果是false则说明只是针对于这个Channel的。

这种数量的设置，也为我们在多个客户端监控同一个queue的这种负载均衡环境下提供了更多的选择。

# 2、常见问题处理 #

## 2.1消息阻塞处理常见解决方案： ##

解决办法应该是在finally语句中来执行这些操作，我分析从队列中取出消息后，会有三种处理结果：

1、处理成功，这种时候应该用basicAck确认消息；

2、可重试的处理失败，这时候应该用basicNack将消息重新入列；

3、不可重试的处理失败，这时候应该使用basicNack将消息丢弃。

常见处理代码如下：

状态定义：

enum
       
        Action { ACCEPT, 
       
        //
       
        处理成功
       
        
       
        RETRY, 
       
        //
       
        可以重试的错误
       
        
       
        REJECT, 
       
        //
       
        无需重试的错误
       
        
       
        } 代码框架如下： Action action 
       
        =
       
        Action.RETRY; 
       
        try
       
        { 
       
        //
       
        如果成功完成则action=Action.ACCEPT
       
        
       
        } 
       
        catch
       
        (Exception e) { 
       
        //
       
        根据异常种类决定是ACCEPT、RETRY还是 REJECT
       
        
       
        } 
       
        finally
       
        { 
       
        //
       
        通过finally块来保证Ack/Nack会且只会执行一次
       
        
       
        
       
        if
       
        (action 
       
        ==
       
        Action.ACCEPT) { channel.basicAck(tag); } 
       
        else
       
        
       
        if
       
        (action 
       
        ==
       
        Action.RETRY) { channel.basicNack(tag, 
       
        false
       
        , 
       
        true
       
        ); } 
       
        else
       
        { channel.basicNack(tag, 
       
        false
       
        , 
       
        false
       
        ); } }

问题：如果重新入队列拿到后还无法处理，怎么办？要控制重入的次数，超过对应的次数业务上要走人工干预处理流程。

## 2.2如何保证消息的顺序性？ ##

为了保障数据的实时性，是多线程处理，每一条数据根据主键ID取模，保障每条被同一个线程处理。线程要设置一个属性表示自己对应的处理那个映射的数据，涉及到分布式环境，需要有一个统一的配置，可以放在zk上进行配置，也可以根据各个节点的名称，按照一定个规则生成，保证不重复，但是如果要保证高可用行，使用zk比较好。如果一个节点发生失败，另为一个节点上去会将对应的任务接手过来进行处理。

## 2.3如何保证从队列接收的程序优雅关闭？ ##

使用jvm hook，挂构子进行处理，接收到kill信号的时候，停止从队列上获取数据，但可以写数据，等待数据处理完成后hook退出，保证接收到的在队列的数据处理完。

# 3、参考文档： #

RabbitMQ使用不当导致的队列堵塞问题及解决之道  [ http://blog.sina.com.cn/s/blog\_48d4cf2d0102w53t.html][http_blog.sina.com.cn_s_blog_48d4cf2d0102w53t.html]

消息的传输控制   [ http://www.cnblogs.com/haoxinyue/archive/2012/10/01/2709644.html][http_www.cnblogs.com_haoxinyue_archive_2012_10_01_2709644.html]

蘑菇街交易平台 数据库架构演进历程  [ http://sanwen8.cn/p/3334qPr.html][http_sanwen8.cn_p_3334qPr.html]

[http_blog.sina.com.cn_s_blog_48d4cf2d0102w53t.html]: http://blog.sina.com.cn/s/blog_48d4cf2d0102w53t.html
[http_www.cnblogs.com_haoxinyue_archive_2012_10_01_2709644.html]: http://www.cnblogs.com/haoxinyue/archive/2012/10/01/2709644.html
[http_sanwen8.cn_p_3334qPr.html]: http://sanwen8.cn/p/3334qPr.html