03 MQ怎么保证消息不丢失

古城微笑少年丶 2022-11-26 01:57 178阅读 0赞

#### 检测消息是否丢失的办法 ####

*  可以利用消息队列的有序性来验证是否有消息丢失。Producer端在发消息时给每个消息附加一个连续递增的序号，然后在Consumer端来检查这个序号的连续性来判断是否有消息丢失，还可以通过缺失的序号来判断丢的是哪条消息。
 *  但是要注意的是像Kafka和RocketMQ这样的消息队列，是不保证在topic上的有序的，只能保证分区上是有序的，所以再发消息时必须指定分区并按区分别编号，并且检测时也要分区单独检测消息的连续性。

#### RocketMQ如何保证消息不丢失的？如何保证可靠性的？ ####

一条消息从生产到被消费，会经过三个阶段：生产阶段、存储阶段、消费阶段。要从这三个阶段保证消息不丢失。  
生产阶段：Producer创建消息，消息经过网络传输发送到Broker端。  
存储阶段：Broker将消息存储在磁盘中。如果是集群的话，消息还会被复制到其他节点上。  
消费阶段：Consumer从Broker上拉取消息，消息经过网络传输发送到Consumer上。

*  **生产阶段**：在生产阶段，是通过请求确认机制，来保证消息的可靠传递的。Producer发送消息给broker，当broker收到后会返回确认信息给Producer。所以生产者通过Broker返回的确认信息，判断消息在生产阶段没有丢失，如果丢失可以重发。（注意要正确处理返回值或者捕获异常，同步发送时，只要注意捕获异常即可；异步发送时，需要在回调方法里检查）
 *  **存储阶段**：在存储阶段，消息到了Broker端，会优先保存在内存中，然后立刻返回确认信息给生产者。随后Broker会定期批量地将消息从内存异步刷到磁盘。这种方式减少了IO次数，性能更好，但是如果Broker宕机，消息还未来得及刷入磁盘，就会出现消息丢失。
    
    如果对消息的可靠性要求非常高，可以通过配置Broker参数来避免因为宕机而丢消息。
    
    1.  首先需要配置Broker参数flushDiskType=SYNC\_FLUSH，意思是将消息保存机制修改为同步刷盘方式，也就是Broker收到消息后，就将消息存储到磁盘，存储成功再给生产者返回确认信息。
    2.  （主从部署）为了保证可用性，Broker通常采用一主多从部署方式，为了保证消息不丢失，还要将消息复制到slave节点。（默认方式下，消息写入master成功，就会返回确认信息给生产者，接着异步将消息复制到slave节点）此时如果master突然宕机不可恢复，那么还未复制到slave的消息就会丢失。为了防止消息丢失，可以采用同步复制方式，master节点把消息成功复制到slave节点之后，才会返回确认响应。需要配置参数brokerRole=SYNC\_MASTER。并且从节点也要设置为同步刷盘。
    
    以上配置虽然可以保证消息不丢失，但是会降低性能，生产中要综合选择。
 *  **消费阶段**：消费者从Broker中拉取消息进行消费，消费成功会发送确认响应给Broker，如果Broker未收到消费确认响应，消费者下次还会再次拉取到该消息，进行重试。

#### Kafka如何保证消息不丢失的？如何保证可靠性的？ ####

*  **生产阶段**：生产者发送消息给Broker后，通过Broker返回的信息来判断消息发送是否成功，如果失败的话就重新发送。  
    ① 使用send方法后，接着调用get方法获取调用结果，但是这样会让它变成同步操作。（不推荐）  
    ② 为send方法添加回调函数来获取结果，  
    要将Producer的retries重试次数设置大一些，可以保证出现网络问题的话可以自动重发消息，避免消息丢失。
 *  **消费阶段**：在分区中有一个offset的概念，表示消费者消费到的位置，默认配置下，当消费者拉取消息之后，消费者会自动提交offset。但是这样可能会产生这样一个问题，消费者刚拿到消息正准备消费时突然挂掉了，消息实际上没有被消费，但是offset却被提交了，就会导致消息丢失。  
    解决办法是关闭自动提交offset，在真正消费完消息之后再手动提交offset。（可能会带来重复消费问题）
 *  **存储阶段**：Kafka为分区引入了多副本机制，多个副本中有一个leader，其他的都是follower。消息会被先发送到leader副本，然后follower副本才能从leader副本中拉取消息进行同步。生产者和消费者只和leader副本交互，其他副本的存在只是为了保证消息存储的安全性。如果leader副本所在的broker挂掉，那么就要从follower副本中重新选出一个leader，但是如果leader的数据还有一部分没有被follower副本同步的话，就会造成消息丢失。

具体参数设置：  
①不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。记住，一 定要使用带有回调通知的 send 方法。  
②设置 retries 为一个较大的值。这里的 retries 同样是 Producer 的参数，对应前面提到 的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。  
③设置 acks = all。acks 是 Producer 的一个参数，代表了你对“已提交”消息的定义。 acks的默认值为1，为1代表消息被leader副本接收就算作被成功发送。配置为all代表所有的副本都接受这个消息之后消息才算被成功发送。  
④设置replication.factor>=3，为了保证数据安全性，一般会保证每个分区至少有3个副本。  
⑤设置min.insync.replicas>1，代表消息至少要被写入2个副本才算被发送成功。min.insync.replicas的默认值为1，在实际生产中尽量避免默认值为1。  
⑥确保 replication.factor > min.insync.replicas。如果两者相等，那么只要有一个副本挂 机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要 在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas + 1。  
⑦设置 unclean.leader.election.enable = false。这是 Broker 端的参数，意思是如果一个 Broker 落后原先的 Leader 太多，就不允许被设置为新的 Leader。  
⑧确保消息消费完成再提交。Consumer 端有个参数 enable.auto.commit，最好把它设 置成 false，并采用手动提交位移的方式。就像前面说的，这对于单 Consumer 多线程 处理的场景而言是至关重要的。