登录 |  注册
首页 >  面试合集 >  Java面试宝典(第三部分·高级) >  解决RabbitMQ消息丢失问题

解决RabbitMQ消息丢失问题

RabbitMQ 消息丢失场景有哪些?

首先我们看下消息周期投递过程:

java.jpg

我们把该图分三部分,左中右,每部分都会导致消息丢失情况,下面就详细聊聊每个阶段消息是如何丢的:

1.生产者生产消息到RabbitMQ Server 消息丢失场景

1) 外界环境问题导致:发生网络丢包、网络故障等造成RabbitMQ Server端收不到消息,因为生产环境的网络是很复杂的,网络抖动,丢包现象很常见,下面会讲到针对这个问题是如何解决的。

2) 代码层面,配置层面,考虑不全导致消息丢失

事例1:

一般情况下,生产者使用Confirm模式投递消息,如果方案不够严谨,比如RabbitMQ Server 接收消息失败后会发送nack消息通知生产者,生产者监听消息失败或者没做任何事情,消息存在丢失风险;

事例2:

生产者发送消息到exchange后,发送的路由和queue没有绑定,消息会存在丢失情况,下面会讲到具体的例子,保证意外情况的发生,即使发生,也在可控范围内。

2.RabbitMQ Server中存储的消息丢失或可靠性不足

1)消息未完全持久化,当机器重启后,消息会全部丢失,甚至Queue也不见了

假如:你仅仅持久化了Message,而Exchange,Queue没有持久化,这个持久化是无效的。记得之前公司有一哥们忘记持久化Queue导致机器重启后,Queue不见了,自然Message也丢失了。

2)单节点模式问题,如果某个节点挂了,消息就不能用了,业务可能瘫痪,只能等待

如果做了消息持久化方案,消息会持久化硬盘,机器重启后消息不会丢失;但是还有一个极端情况,这台服务器磁盘突然坏了(公司遇到过磁盘问题还是很多的),消息持久化不了,非高可用状态,这个模式生产环境慎重考虑。

3)普通集群模式:某个节点挂了,该节点上的消息不能用,有影响的业务瘫痪,只能等待节点恢复重启可用(建立在消息持久化)

虽然这个模式进步了一点点,多个节点,但是消息还是不能保证可靠,为什么呢?

因为RabbitMQ 集群模式有点特殊,队列的内容仅仅存在某一个节点上面,不会存在所有节点上面,所有节点仅仅存放消息结构和元数据(可以理解为索引,这也是为了提高性能,如果每次把所有内容同步到所有节点是有开销代价的)。下面自己画了一张图介绍普通集群丢失消息情况:

java.jpg

这里有三个节点,通常情况下一个磁盘节点,两个内存节点,首先先说明下, Queue1 内容仅仅存在节点note1上面,在创建队列的时候已经固定了,note2,note3 仅仅存放的是元数据,这个一定要清楚,Producer发送消息到note2,note2 会同步元数据到其他节点,内容会同步note1。

那我们想下,图中的Q1问题,note1挂了,这个节点的Queues全部暂时不可用,节点恢复后可用。

我们说下图片中备注2中的问题,Producer发送消息到note2,note2在同步note1前note1挂了,此时你的心情是怎么样的。。。后面会讲具体的策略

4)镜像模式:可以解决上面的问题,但是还是有意外情况发生

比如:持久化的消息,保存到硬盘过程中,当前队列节点挂了,存储节点硬盘又坏了,消息丢了,怎么办?下面会详细介绍

3.RabbitMQ Server到消费者消息丢失

消费端接收到相关消息之后,消费端还没来得及处理消息,消费端机器就宕机了,此时消息如果处理不当会有丢失风险,后面会讲到如何处理这个情况,消费端也有ack机制

如何避免消息丢失?

下面也是从三个方面介绍:

1.生产者生产消息到RabbitMQ Server 可靠性保证

2.RabbitMQ Server中存储的消息如何保证

3.RabbitMQ Server到消费者消息如何不丢

1. 生产者生产消息到RabbitMQ Server可靠性保证

这个过程,消息可能会丢,比如发生网络丢包、网络故障等造成消息丢失,一般情况下如果不采取措施,生产者无法感知消息是否已经正确无误的发送到exchange中,如果生产者能感知到的话,它可以进行进一步的处理动作,比如重新投递相关消息以确保消息的可靠性。

1.1 通常有一种方案可以解决:就是 AMQP协议提供的一个事务机制

RabbitMQ客户端中Channel 接口提供了几个事务机制相关的方法:

channel.txSelect

channel.txCommit

channel.txRollback

源码截图如下:com.rabbitmq.client 包中public interface Channel extendsShutdownNotifier {}接口

java.jpg

在生产者发送消息之前,通过channel.txSelect开启一个事务,接着发送消息,

如果消息投递server失败,进行事务回滚channel.txRollback,然后重新发送,

如果server收到消息,就提交事务channel.txCommit

但是,很少有人这么干,因为这是同步操作,一条消息发送之后会使发送端阻塞,以等待RabbitMQ Server的回应,之后才能继续发送下一条消息,生产者生产消息的吞吐量和性能都会大大降低。

1.2 幸运的是RabbitMQ提供了一个改进方案,即发送方确认机制(publisher confirm)

首先生产者通过调用channel.confirmSelect方法将信道设置为confirm模式,一旦信道进入confirm模式,所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始),一旦消息被投递到所有匹配的队列之后,RabbitMQ就会发送一个确认(Basic.Ack)给生产者(包含消息的唯一deliveryTag和multiple参数),这就使得生产者知晓消息已经正确到达了目的地了。

其实Confirm模式有三种方式实现:

  • 串行confirm模式:producer每发送一条消息后,调用waitForConfirms()方法,等待broker端confirm,如果服务器端返回false或者在超时时间内未返回,客户端进行消息重传。

  • 批量confirm模式:producer每发送一批消息后,调用waitForConfirms()方法,等待broker端confirm。

  • 异步confirm模式:提供一个回调方法,broker confirm了一条或者多条消息后producer端会回调这个方法。

我们分别来看看这三种confirm模式

串行confirm

for(int i = 0;i<50;i++){
    channel.basicPublish(
           exchange, routingKey,
            mandatory, immediate,
            messageProperties,
            message.getContent()
    );    
if (channel.waitForConfirms()) {
        System.out.println("发送成功");    
} else {        
//发送失败这里可进行消息重新投递的逻辑        
System.out.println("发送失败");
    }
}

批量confirm模式

for(int i = 0;i<50;i++){
    channel.basicPublish(
            exchange, routingKey,
            mandatory, immediate,
            messageProperties,
            message.getContent()
    );
}
if (channel.waitForConfirms())
{
    System.out.println("发送成功");
} else {    
System.out.println("发送失败");
}

上面代码是简单版本的,生产环境绝对不是循环发送的,而是根据业务情况,

各个客户端程序需要定期(每x秒)或定量(每x条)或者两者结合来publish消息,然后等待服务器端confirm。相比普通confirm模式,批量可以极大提升confirm效率。

但是有没有发现什么问题?

问题1: 批量发送的逻辑复杂化了。

问题2: 一旦出现confirm返回false或者超时的情况时,客户端需要将这一批次的消息全部重发,这会带来明显的重复消息数量,并且当消息经常丢失时,批量confirm性能应该是不升反降的。

异步confirm模式

Channel channel = channelManager.getPublisherChannel(namespaceName);
ProxiedConfirmListener confirmListener = new ProxiedConfirmListener();
//监听类
confirmListener.setChannelManager(channelManager);
confirmListener.setChannel(channel);
confirmListener.setNamespace(namespaceName);
confirmListener.addSuccessCallbacks(successCallbacks);
channel.addConfirmListener(confirmListener);
channel.confirmSelect();
//开启confirm模式
AMQP.BasicProperties messageProperties = null;
if (message.getProperty() instanceof AMQP.BasicProperties) {
    messageProperties = (AMQP.BasicProperties) message.getProperty();
}
confirmListener.toConfirm(channel.getNextPublishSeqNo(), rawMsg);
for(int i = 0;i<50;i++){
    channel.basicPublish(
            exchange, routingKey,
            mandatory, immediate,
            messageProperties,
            message.getContent()
    );
}

异步模式需要自己多写一部分复杂的代码实现,异步监听类,监听server端的通知消息,异步的好处性能会大幅度提升,发送完毕之后,可以继续发送其他消息。MQServer通知生产端ConfirmListener监听类:用户可以继承接口实现自己的实现类,处理消息确认机制,此处继承类代码省略,就是上面 ProxiedConfirmListener 类:下面贴下要实现的接口:

package com.rabbitmq.client;
import java.io.IOException;
/**
 * Implement this interface in order to be notified of Confirm events.
 * Acks represent messages handled successfully; Nacks represent
 * messages lost by the broker.  Note, the lost messages could still
 * have been delivered to consumers, but the broker cannot guarantee
 * this.
 */
public interface ConfirmListener {
    /**
    ** handleAck RabbitMQ消息接收成功的方法,成功后业务可以做的事情
    ** 发送端投递消息前,需要把消息先存起来,比如用KV存储,接收到ack后删除
    **/
    void handleAck(long deliveryTag, boolean multiple)    throws IOException;
    //handleNack RabbitMQ消息接收失败的通知方法,用户可以在这里重新投递消息
    void handleNack(long deliveryTag, boolean multiple)        throws IOException;
}

上面的接口很有意思,如果是你的话,怎么实现?消息投递前如何存储消息,ack 和 nack 如何处理消息?

下面看下异步confirm的消息投递流程:

java.jpg

解释下这张图片:

channel1 连续发类1,2,3条消息到RabbitMQ-Server,RabbitMQ-Server通知返回一条通知,里面包含回传给生产者的确认消息中的deliveryTag包含了确认消息的序号,此外还有一个参数multiple=true,表示到这个序号之前的所有消息都已经得到了处理。这样客户端和服务端通知的次数就减少类,提升类性能。

channel3 发送的消息失败了,生产端需要对投递消息重新投递,需要额外处理代码。那么生产端需要做什么事情呢?因为是异步的,生产端需要存储消息然后根据server通知的消息,确认如何处理,于是我们面临的问题是:

第一:发送消息之前把消息存起来

第二:监听ack 和 nack 并做响应处理

那么怎么存储呢?

我们分析下,可以使用SortedMap 存储,保证有序,但是有个问题高并发情况下,

每秒可能几千甚至上万的消息投递出去,消息的ack要等几百毫秒的话,放内存可能有内存溢出的风险。所以建议采用KV存储,KV存储承载高并发能力高,性能好,但是要保证KV 高可用,单个有个缺点就是又引入了第三方中间件,复杂度升高。

解决了上面的问题,下面还会遇到一个问题,消息丢失的另一个情况?

事务机制和publisher confirm机制确保的是消息能够正确的发送至RabbitMQ,这里的“发送至RabbitMQ”的含义是指消息被正确的发往至RabbitMQ的交换器,如果此交换器没有匹配的队列的话,那么消息也将会丢失,怎么办?

1. 使用mandatory 设置true

2. 利用备份交换机(alternate-exchange):实现没有路由到队列的消息

我们看下RabbitMQ客户端代码方法

Channel 类中 发布消息方法

void basicPublish ( String exchange , String routingKey , boolean mandatory , boolean immediate , BasicProperties props , byte [] body ) throws IOException ;

解释下:basicPublish 方法中的,mandatory和immediate

/**
     * 当mandatory标志位设置为true时,如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue, 那么会调用basic.return方法将消息返回给生产者<br>
     * 当mandaory设置为false时,出现上述情形broker会直接将消息扔掉。
     */
@Setter ( AccessLevel . PACKAGE ) private boolean mandatory = false ;    
/**
     * 当immediate标志位设置为true时,如果exchange在将消息路由到queue(s)时发现对于的queue上没有消费者, 那么这条消息不会放入队列中。
     当immediate标志位设置为false时,exchange路由的队列没有消费者时,该消息会通过basic.return方法返还给生产者。
     * RabbitMQ 3.0版本开始去掉了对于immediate参数的支持,对此RabbitMQ官方解释是:这个关键字违背了生产者和消费者之间解耦的特性,因为生产者不关心消息是否被消费者消费掉
     */
@Setter ( AccessLevel . PACKAGE ) private boolean immediate ;

    所以为了保证消息的可靠性,需要设置发送消息代码逻辑。如果不单独形式设置mandatory=false

使用mandatory 设置true的时候有个关键点要调整,生产者如何获取到没有被正确路由到合适队列的消息呢?通过调用channel.addReturnListener来添加ReturnListener监听器实现,只要发送的消息,没有路由到具体的队列,ReturnListener就会收到监听消息。

上一篇: RabbitMQ 如何保证消息的可靠性
下一篇: RabbitMQ 持久化成功的条件
推荐文章
  • 在HTML中,如果你想让一个输入框(input元素)不可编辑,你可以通过设置其readonly属性来实现。示例如下:input type="text" value="此处内容不可编辑" readonly在上述代码中,readonly属性使得用户无法修改输入框中的内容。另外,如果你希望输入框完全不可交
  • ASP.NET教程ASP.NET又称为ASP+,基于.NETFramework的Web开发平台,是微软公司推出的新一代脚本语言。ASP.NET是一个使用HTML、CSS、JavaScript和服务器脚本创建网页和网站的开发框架。ASP.NET支持三种不一样的开发模式:WebPages(Web页面)、
  • C# 判断判断结构要求程序员指定一个或多个要评估或测试的条件,以及条件为真时要执行的语句(必需的)和条件为假时要执行的语句(可选的)。下面是大多数编程语言中典型的判断结构的通常形式:判断语句C#提供了以下类型的判断语句。点击链接查看每个语句的细节。语句描述if语句一个 if语句 由一个布尔表达式后跟
  • C#循环有的时候,可能需要多次执行同一块代码。通常情况下,语句是顺序执行的:函数中的第一个语句先执行,接着是第二个语句,依此类推。编程语言提供了允许更为复杂的执行路径的多种控制结构。循环语句允许我们多次执行一个语句或语句组,下面是大多数编程语言中循环语句的通常形式:循环类型C#提供了以下几种循环类型
  • C#数组(Array)数组是一个存储相同类型元素的固定大小的顺序集合。数组是用来存储数据的集合,一般认为数组是一个同一类型变量的集合。声明数组变量并不是声明number0、number1、...、number99一个个单独的变量,而是声明一个就像numbers这样的变量,然后使用numbers[0]
  • ASP.NET是一个由微软公司开发的用于构建Web应用程序的框架,它是.NETFramework的一部分。它提供了一种模型-视图-控制器(MVC)架构、Web表单以及最新的ASP.NETCore中的RazorPages等多种开发模式,可以用来创建动态网页和Web服务。以下是一些基础的ASP.NET编
学习大纲