RocketMQ系列 | 全网最全的导致RocketMQ消息“丢失”的几个场景都在这了，肯定有你不知道！

2024-01-13 884

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 发送时会丢失消息、消息存储场景丢失消息、消费时会丢失消息

RocketMQ 简介

RocketMQ 5.0：
云原生“消息、事件、流”实时数据处理平台，覆盖云边端一体化数据处理场景。

RocketMQ领域模型

如上图所示，Apache RocketMQ 中消息的生命周期主要分为消息生产、消息存储、消息消费这三部分。

生产者生产消息并发送至 Apache RocketMQ 服务端，消息被存储在服务端的主题[Topic]中，消费者通过订阅主题[Topic]消费消息。

消息生产

生产者（Producer）：Apache RocketMQ 领域中用于产生消息的运行实体，一般集成于业务调用链路的上游。

消息存储

主题（Topic）：Apache RocketMQ 消息传输和存储的分组容器，主题内部由多个队列组成，消息的存储和水平扩展实际是通过主题内的队列实现的。
队列（MessageQueue）：Apache RocketMQ 消息传输和存储的实际单元容器，类比于其他消息队列中的分区。Apache RocketMQ 通过流式特性的无限队列结构来存储消息，消息在队列内具备顺序性存储特征。
消息（Message）：Apache RocketMQ 的最小传输单元。消息具备不可变性，在初始化发送和完成存储后即不可变。

消息消费

消费者分组（ConsumerGroup）：Apache RocketMQ 发布订阅模型中定义的独立的消费身份分组，用于统一管理底层运行的多个消费者（Consumer）。同一个消费组的多个消费者必须保持消费逻辑和配置一致，共同分担该消费组订阅的消息，实现消费能力的水平扩展。
消费者（Consumer）：Apache RocketMQ 消费消息的运行实体，一般集成在业务调用链路的下游。消费者必须被指定到某一个消费组中。
订阅关系（Subscription）：Apache RocketMQ 发布订阅模型中消息过滤、重试、消费进度的规则配置。订阅关系以消费组粒度进行管理，消费组通过定义订阅关系控制指定消费组下的消费者如何实现消息过滤、消费重试及消费进度恢复等。Apache RocketMQ 的订阅关系除过滤表达式之外都是持久化的，即服务端重启或请求断开，订阅关系依然保留。

如何让“消息丢失”？

在“如何让消息丢失”之前，让我们梳理一下消息的生命周期，先对齐下整体的概念。

一条消息的历程

1、发送场景丢失消息

1.1 单向发送

/**     * 发送消息，Oneway形式，服务器不应答，     * 无法保证消息是否成功到达服务器     *     * @param message 要发送的消息     */voidsendOneway(finalMessagemessage);com.aliyun.openservices.ons.api.Producer#sendOneway

RocketMQ 提供三种方式来发送普通消息：同步（Sync）发送、异步（Async）发送和单向（Oneway）发送。

同步发送：同步发送是指消息发送方发出一条消息后，会在收到服务端同步响应之后才发下一条消息的通讯方式。

此种方式应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。

异步发送：异步发送是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。消息发送方在发送了一条消息后，不需要等待服务端响应即可发送第二条消息。发送方通过回调接口接收服务端响应，并处理响应结果。

一般用于链路耗时较长，对响应时间较为敏感的业务场景，例如，您视频上传后通知启动转码服务，转码完成后通知推送转码结果等。

单向发送：发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不等待应答。

此方式发送消息的过程耗时非常短，一般在微秒级别。适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。

1.2 发送失败时未重试或补偿

importcom.aliyun.openservices.ons.api.Message;importcom.aliyun.openservices.ons.api.Producer;importcom.aliyun.openservices.ons.api.SendResult;importlombok.extern.slf4j.Slf4j;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Service;
importjava.nio.charset.StandardCharsets;
@Service@Slf4jpublicclassProductSender {
@AutowiredprivateProducerproducer;
publicvoidsendMsg(Stringcontent) {        try {            byte[] body=content.getBytes(StandardCharsets.UTF_8);            Messagemessage=newMessage("topicName", "tagName", "msgKey", body);            SendResultsendResult=producer.send(message);        } catch (Exceptionignored) {            // TODO: 2023/9/29 发送失败时无处理，网络抖动或服务不稳定时会造成消息丢失        }    }}

2、消息存储场景丢失消息

2.1 、Broker宕机或者磁盘损坏，Broker Server内存中的消息没有落盘

2.2 、过期清理机制引发消息丢失
Apache RocketMQ 中队列的定义，消息按照到达服务器的先后顺序被存储到队列中，理论上每个队列都支持无限存储。但是在实际部署场景中，服务端节点的物理存储空间有限，消息无法做到永久存储。因此，在实际使用中需要考虑以下问题，消息在服务端中的存储以什么维度为判定条件？消息存储以什么粒度进行管理？消息存储超过限制后如何处理？这些问题都是由消息存储和过期清理机制来定义的。

Apache RocketMQ 使用存储时长作为消息存储的依据，即每个节点对外承诺消息的存储时长。在存储时长范围内的消息都会被保留，无论消息是否被消费；超过时长限制的消息则会被清理掉。删除旧的没有使用过的消息是由后台定时任务完成的。

消息存储文件结构说明

3、消费场景丢失消息

3.1 消费失败，但消费消息的返回结果为成功

importcom.aliyun.openservices.ons.api.Action;importcom.aliyun.openservices.ons.api.ConsumeContext;importcom.aliyun.openservices.ons.api.Message;importcom.aliyun.openservices.ons.api.MessageListener;importlombok.extern.slf4j.Slf4j;importorg.springframework.stereotype.Service;
importjava.nio.charset.StandardCharsets;
@Service@Slf4jpublicclassMissingMsgWhenConsumeFailimplementsMessageListener {
@OverridepublicActionconsume(Messagemessage, ConsumeContextcontext) {        
try {            Stringmsg=newString(message.getBody(), StandardCharsets.UTF_8);            returnAction.CommitMessage;        } catch (Exceptione) {            //丢失消息：消费失败了，但消费消息的返回结果为成功。            return Action.CommitMessage;        }    }}