RocketMQ消息丢失的场景以及解决方案-六虎

引言

Apache RocketMQ是企业级的音讯中间件，以其高功能和高可靠性而广泛应用。可是，音讯丢掉的问题在实践中依然存在，本文将探讨此问题并供给解决计划。

1. 音讯发送环节

事例：网络毛病导致的音讯发送失利
在Spring Boot应用中，生产者或许会遇到网络瞬断，导致音讯未成功发送到Broker。

代码示例：

@Service
public class ProducerService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    public void sendMessage(String topic, String message) {
        try {
            rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(message).build());
        } catch (MQClientException e) {
            // 反常处理逻辑，例如记载日志、报警等
        }
    }
}

解决计划：
经过装备生产者的重试次数，咱们能够强化音讯的发送可靠性。

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class ProducerService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    public void configureProducerRetry() {
        // 获取RocketMQ的生产者客户端
        DefaultMQProducer producer = rocketMQTemplate.getProducer();
        // 设置发送失利时的重试次数为5
        producer.setRetryTimesWhenSendFailed(5);
        // 设置发送失利时重试另一个Broker
        producer.setRetryAnotherBrokerWhenNotStoreOK(true);
        // 其他生产者装备...
    }
    public void sendMessage(String topic, String message) {
        // 在发送前保证调用了装备重试的办法
        configureProducerRetry();
        // 发送音讯
        rocketMQTemplate.syncSend(topic, message);
    }
}

2. Broker存储环节

事例：Broker宕机导致的音讯存储失利
Broker在接纳音讯后，在持久化之前产生毛病，这会导致音讯丢掉。

解决计划：

同步刷盘装备

在RocketMQ中，同步刷盘是指Broker在回来音讯发送成功之前，将音讯持久化到磁盘。这样做能够保证即便Broker产生毛病，音讯也不会丢掉。

要启用同步刷盘，需求修改Broker的装备文件，一般是broker.properties文件，设置如下特点：

flushDiskType=SYNC_FLUSH

当flushDiskType设置为SYNC_FLUSH时，每次音讯接纳后，Broker都会同步地将音讯写入磁盘中。这保证了音讯的持久性，但或许会对功能产生影响，因为每次音讯写入都需求磁盘IO操作。

副本机制

RocketMQ运用主从架构来供给数据的高可用性。主Broker负责处理音讯的读写恳求，而从Broker则仿制主Broker的数据。假如主Broker不可用，从Broker能够接纳作业，保证音讯不会丢掉。

要装备副本机制，能够在布置RocketMQ集群时，为每个Master Broker设置一个或多个Slave Broker。在Broker的装备文件中，设置如下特点：

brokerRole=SYNC_MASTER  # 关于主Broker
brokerRole=SLAVE        # 关于从Broker

此外，还需求在称号服务器（NameServer）装备中指定一切Broker的地址，以便生产者和顾客能够发现它们。

留意：副本数量的添加需求在RocketMQ集群布置时进行规划，需求考虑到资源耗费和数据一致性的要求。从Broker不会对外供给服务，它的角色首要是数据的同步和在主Broker不可用时的毛病转移。

关于生产环境，主张进行充沛的测验，以平衡功能和可靠性的需求。正确装备同步刷盘和副本机制，能够极大地增强RocketMQ的音讯可靠性。一起，这些装备一般需求和其他体系资源（如磁盘功能、网络带宽等）一起考虑，以保证整体的体系安稳性和功能。

3. 音讯消费环节

事例：顾客反常导致的音讯消费失利
顾客在处理音讯时产生反常，比方数据库操作失利，导致音讯消费不成功。

代码示例：

@RocketMQMessageListener(topic = "order-topic", consumerGroup = "order-consumer-group")
public class OrderConsumer implements RocketMQListener<Order> {
    @Override
    public void onMessage(Order order) {
        try {
            // 处理订单逻辑
        } catch (Exception e) {
            // 反常处理逻辑，如重试或记载失利的音讯
        }
    }
}

解决计划：
能够在顾客中完成逻辑保证音讯在消费成功后，再发送承认。

在RocketMQ中，手动音讯承认机制是指顾客在成功处理完音讯后，需求显式地发送一个承认（acknowledgment）回Broker，告知它音讯现已被成功消费。这样做的目的是为了保证音讯不会因为顾客的毛病而丢掉，一起防止音讯被重复处理。

手动承认形式一般用于保证音讯传递的可靠性，特别是在需求保证音讯被准确一次处理的场景下。

以下是运用手动承认办法的完好计划：

装备顾客运用手动承认形式：
在@RocketMQMessageListener注解中设置messageModel参数为MessageModel.CLUSTERING和consumeMode为ConsumeMode.ORDERLY或ConsumeMode.CONCURRENTLY，取决于是否需求保证音讯顺序。
处理音讯：
完成RocketMQPushConsumerLifecycleListener接口，并在prepareStart办法中注册MessageListener。在MessageListener的完成中，处理音讯并回来相应的消费状况。
承认音讯：
假如音讯成功处理，回来ConsumeConcurrentlyStatus.CONSUME_SUCCESS或ConsumeOrderlyStatus.SUCCESS状况，这将会告知Broker音讯现已被消费，能够从行列中移除。

假如处理音讯时产生反常或需求稍后从头消费，回来ConsumeConcurrentlyStatus.RECONSUME_LATER或ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT，Broker将会稍后从头发送该音讯。
反常处理：
假如在消费过程中产生反常，能够捕获反常并记载必要的日志，然后挑选是当即重试还是推迟重试。

以下是一个示例代码：

@RocketMQMessageListener(topic = "order-topic", consumerGroup = "order-consumer-group",consumeMode = ConsumeMode.ORDERLY)
public class OrderConsumer implements RocketMQPushConsumerLifecycleListener {
    @Override
    public void prepareStart(final DefaultMQPushConsumer consumer) {
        // 设置顾客其他特点...
        // 设置音讯监听器
        consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
            MessageExt msg = msgs.get(0); // 假设一次只消费一条音讯
            try {
                // 处理音讯
                // ...
                // 假如音讯处理成功，承认音讯
                return ConsumeOrderlyStatus.SUCCESS;
            } catch (Exception e) {
                // 处理反常
                // ...
                // 假如需求稍后从头消费音讯
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        });
    }
}

在这个比如中，OrderConsumer类完成了RocketMQPushConsumerLifecycleListener接口，并在prepareStart办法中注册了一个MessageListenerOrderly，以保证音讯以有序的办法被消费。根据音讯处理结果，它回来相应的状况码，然后完成手动承认。

留意：在RocketMQ中，除了手动承认外，还有主动承认机制。在主动承认形式下，当顾客从Broker拉取到音讯并由客户端代理（即SDK）接纳后，假如没有产生反常，音讯会主动被承认消费成功。这种形式适用于那些对音讯处理的可靠性要求不是十分高的场景。

在主动承认机制中，顾客无需编写额外的承认逻辑。假如顾客在处理音讯时没有抛出反常，SDK会主意向Broker发送ACK（承认）。假如处理过程中抛出了反常，音讯会根据设定的重试战略再次发送给顾客。

默认情况下，RocketMQ的顾客采用的是主动承认机制。这意味着一旦顾客监听器办法执行完毕，不管其结果如何，音讯都会被标记为已消费。假如在消费过程中呈现了反常，RocketMQ客户端会根据装备的重试战略来从头投递音讯。
这种主动承认机制简化了代码，可是假如在音讯处理过程中需求更细粒度的控制，或许需求保证音讯即便在消费过程中呈现反常也不会丢掉，那么应该运用手动承认机制。

4. 高可用性问题

事例：主从同步推迟导致的音讯丢掉
在Broker主从同步装备不当的情况下，主Broker毛病或许导致音讯丢掉。

解决计划：

为了保证在RocketMQ中音讯的可靠性，特别是在呈现毛病时防止数据丢掉，能够采取以下的解决计划：

1. 同步主从装备（SYNC_MASTER）

在RocketMQ中，SYNC_MASTER是一个高可用性设置，它要求每条音讯在承认给生产者之前，不仅在主Broker上写入磁盘，还要同步到一切的从Broker。这保证了即便主Broker呈现毛病，音讯也不会丢掉，因为它现已存在于从Broker中。

为了装备SYNC_MASTER，需求在主Broker的装备文件（一般是broker-a.properties）中设置：

brokerRole=SYNC_MASTER

而且在从Broker的装备文件（例如broker-b.properties或broker-b-s.properties）中设置：

brokerRole=SLAVE

这样装备后，主Broker会等候音讯同步到从Broker后才向生产者承认音讯发送成功。

留意

这种装备办法存在如下问题，首要包含：

功能开销：
同步仿制意味着每条音讯都需求在被承认前写入主Broker并仿制到从Broker。这个过程涉及额外的网络IO和磁盘IO，会添加音讯的推迟时刻，尤其是在高负载情况下。
资源耗费：
因为每条音讯都需求在多个Broker上存储，因而会添加存储资源的运用。此外，同步仿制还会添加网络带宽的占用。
扩展性问题：
当集群规划增大时，同步仿制或许会成为瓶颈。因为一切的从Broker都需求与主Broker坚持实时数据同步，大量的同步操作或许会导致体系扩展性受限。
体系复杂性：
同步主从装备添加了体系的复杂性，需求更多的办理和维护作业。例如，需求保证主从之间的同步机制正常作业，而且当主Broker呈现问题时，从Broker能够及时接纳作业。
高可用性与功能的权衡：
尽管SYNC_MASTER装备能够供给较高的数据可靠性，但功能上的折衷或许使得它不适用于对推迟灵敏或需求极高吞吐量的应用场景。
毛病恢复时刻：
当主Broker失利后，体系需求花费时刻来切换到从Broker，这期间体系的可用性或许会受到影响。

因而，运用SYNC_MASTER装备时需求根据实践事务场景和需求来平衡可靠性和功能，或许需求在可接受的音讯推迟和体系资源运用范围内进行权衡。在不需求严厉音讯顺序的场景下，能够考虑运用异步仿制来进步功能。

2. 数据备份

为了进一步保护数据免于丢掉，能够定时对音讯日志和消费进度信息进行备份。这不仅包含了音讯体自身，还包含了一切的顾客偏移量和行列信息。

备份战略或许包含以下几点：

定时备份：运用脚本或现成的备份工具，定时（如每天）将数据从Broker服务器仿制到一个安全的备份位置。
实时备份：关于极端要害的数据，或许需求考虑更高档的备份计划，例如运用磁盘阵列的镜像功能，或许集成第三方的实时数据备份解决计划。
备份验证：定时对备份进行恢复测验，以验证备份的完好性和有效性。
3. 监控和报警

树立监控体系以监测Broker的状况和功能指标，及时发现并处理同步推迟或失利等问题。一起，应装备报警机制，在检测到或许导致数据丢掉的反常时当即告诉运维人员。

4. 集群布置

在不同的数据中心布置多个Broker集群，以防单一数据中心呈现毛病时影响整个音讯体系。跨数据中心的仿制能够经过RocketMQ的跨站点（cross-site）仿制功能来完成。

经过施行这些战略，能够保证RocketMQ体系在大都毛病情况下都能保证音讯的不丢掉，进步整个音讯体系的健壮性和可靠性。这些办法需求结合详细的事务需求和体系环境来详细施行。

结语

综合以上的战略，咱们能够明显增强在应用中运用RocketMQ时的音讯持久性和可靠性。经过细心装备音讯发送重试机制、同步刷盘、主从同步仿制以及施行定时数据备份和强化监控报警体系，开发者能够为音讯体系构建一个愈加坚固的安全网。然而，抱负的装备计划往往需求根据事务的详细需求和体系的运转环境来定制。主张在布置前进行充沛的测验，以保证体系的安稳性，并在实践运转中持续监控和调整，以应对不断改变的事务和技术环境。这样的实践将大大削减音讯丢掉的或许性，为应用供给安稳可靠的音讯交换保障。

RocketMQ消息丢失的场景以及解决方案

引言

1. 音讯发送环节

2. Broker存储环节

3. 音讯消费环节

4. 高可用性问题

结语

相关文章

从一个线上 Android Bug 回看 Fragment 的基础知识

关于我在学习LFU的时候，在开源项目捡了个漏这件事。

2022年生活总结

关于GPT的Open API，看这一篇就够了（可搭建自己的ChatGPT）

作者信息