引入消息队列可以方便地实现系统解耦、削峰填谷等作用。但是消息队列使用不当，可能会引起消息丢失，在一些消息敏感的业务场景下，这是不允许的。今天我们来聊一聊 rocketmq 怎么做能确保消息不丢失。
一、rocketmq 简介
rocketmq 是阿里巴巴开源的分布式消息中间件，整体架构如下图：
rocketmq 主要包括 producer、consumer 和 broker，同时 name server 进行集群注册管理和保存元数据。
二、消息不丢失
要想保证消息不丢失，需要从以下几个方面考虑：
producer 发送消息。
broker 保存消息。
consumer 消费消息。
broker 主从切换。
1、同步发送
*如下：
public void send() throws exception catch (exception e) if (sendresult !=null) }同步发送会返回 4 个状态码：
send_ok：消息发送成功。需要注意的是，消息发送到 broker 后，还有两个操作：消息刷盘和消息同步到 sl**e 节点，默认这两个操作都是异步的，只有把这两个操作都改为同步，send_ok 这个状态才能真正表示发送成功。
flush_disk_timeout：消息发送成功但是消息刷盘超时。
flush_sl**e_timeout：消息发送成功但是消息同步到 sl**e 节点时超时。
sl**e_not_**ailable：消息发送成功但是 broker 的 sl**e 节点不可用。
根据返回的状态码，可以做消息重试，这里设置的重试次数是 3。
消息重试时，消费端一定要做好幂等处理。
2、异步发送
*如下：
public void sendasync() throws exception @override public void onexception(throwable e)异步发送，可以重写**函数，**函数捕获到 exception 时表示发送失败，这时可以进行重试，这里设置的重试次数是 3。
3、刷盘策略
1）异步刷盘
默认。消息写入 commitlog 时，并不会直接写入磁盘，而是先写入 pagecache 缓存后返回成功，然后用后台线程异步把消息刷入磁盘。异步刷盘提高了消息吞吐量，但是可能会有消息丢失的情况，比如断点导致机器停机，pagecache 中没来得及刷盘的消息就会丢失。
2）同步刷盘
消息写入内存后，立刻请求刷盘线程进行刷盘，如果消息未在约定的时间内(默认 5 s)刷盘成功，就返回 flush_disk_timeout，producer 收到这个响应后，可以进行重试。同步刷盘策略保证了消息的可靠性，同时降低了吞吐量，增加了延迟。要开启同步刷盘，需要增加下面配置：
flushdisktype=sync_flush
4、broker 多副本和高可用
broker 为了保证高可用，采用一主多从的方式部署。如下图：
消息发送到 master 节点后，sl**e 节点会从 master 拉取消息保持跟 master 的一致。这个过程默认是异步的，即 master 收到消息后，不等 sl**e 节点复制消息就直接给 producer 返回成功。
这样会有一个问题，如果 sl**e 节点还没有完成消息复制，这时 master 宕机了，进行主备切换后就会有消息丢失。为了避免这个问题，可以采用 sl**e 节点同步复制消息，即等 sl**e 节点复制消息成功后再给 producer 返回发送成功。只需要增加下面的配置：
brokerrole=sync_master
改为同步复制后，消息复制流程如下：
sl**e 初始化后，跟 master 建立连接并向 master 发送自己的 offset；
master 收到 sl**e 发送的 offset 后，将 offset 后面的消息批量发送给 sl**e；
sl**e 把收到的消息写入 commitlog 文件，并给 master 发送新的 offset；
master 收到新的 offset 后，如果 offset >=producer 发送消息后的 offset，给 producer 返回 send_ok。
5、消息确认
consumer 消费消息的**如下：
public void consume() throws exception catch (exception e) }consumer.start();如果 consumer 消费成功，返回 consume_success，提交 offset 并从 broker 拉取下一批消息。
6、consumer 重试
consumer 消费失败，这里有 3 种情况：
返回 reconsume_later
返回 null
抛出异常。broker 收到这个响应后，会把这条消息放入重试队列，重新发送给 consumer。
注意：
broker 默认最多重试 16 次，如果重试 16 次都失败，就把这条消息放入死信队列，consumer 可以订阅死信队列进行消费。
重试只有在集群模式（messagemodel.clustering）下生效，在广播模式（messagemodel.broadcasting）下是不生效的。
consumer 端一定要做好幂等处理。
其实重试 3 次都失败就可以说明**有问题，这时 consumer 可以把消息存入本地，给 broker 返回consume_success 来结束重试。**如下：
int count = messageext) msgs).getreconsumetimes();if (count > 2)7、事务消息
rocketmq支持事务消息，整体流程如下图：
producer 发送 half 消息；
broker 先把消息写入 topic 是 rmq_sys_trans_half_topic 的队列，之后给 producer 返回成功；
producer 执行本地事务，成功后给 broker 发送 commit 命令（本地事务执行失败则发送 rollback）；
broker 收到 commit 请求后把消息状态更改为成功并把消息推到真正的 topic；
consumer 拉取消息进行消费。
*如下：
public class producertransactionlistenerimpl implements transactionlistener @override public localtransactionstate checklocaltransaction(messageext msg) }8、消息索引
我们知道，rocketmq 核心的数据文件有 3 个：commitlog、consumequeue 和 index。其中index 文件就是一个索引文件，结构如下图：
查找消息时，首先根据消息 key 的 hashcode 计算出 hash 槽的位置，然后读取 hash 槽的值计算 index 条目的位置，从index 条目位置读取到消息在 commitlog 文件中的 offset，从而查找到消息。
在 producer 发送消息时，可以指定一个 key，**如下：
message sendmessage = new message("topic1", tag1", message.getbytes())sendmessage.setkeys("weiyiid");这样可以通过 rocketmq 提供的命令或者管理控制台来查询消息是否发送成功。
9、极端情况
如果对消息丢失零容忍，我们必须要考虑极端情况，比如整个 rocketmq 集群挂了，这时 producer 端发送消息一定会失败，可以考虑在 producer 端做降级，把要发送的消息保存到本地数据库或磁盘，等 rocketmq 恢复以后再把本地消息推送出去。
三、总结
在一些特殊的业务场景，比如支付、银行核算等，需要确保消息不丢失，但是同时也要看到，消息不丢失的方案会大大降低 rocketmq 的吞吐量，需要综合考虑。
作者丨朱晋君**丨***君哥聊技术（id：gh_1f109b82d301）dbaplus社群欢迎广大技术人员投稿，投稿邮箱：[email protected]
9个方面全面解析留学马来西亚申请时间
马来西亚以其优质的教育资源和相对低廉的留学成本吸引着众多国际学生。要想成功申请马来西亚的留学签证，了解申请流程和时间安排非常重要。本文将为您提供一份全面的马来西亚留学申请时间指南。.申请时间安排。留学马来西亚的申请时间一般需要提前半年至一年开始准备。建议在申请截止日期前至少提前两个月开始准备，以防止...
如何查询小学学校代码 全面解析与步骤
在当今的教育信息化时代，学校 已经成为了每所学校的唯一标识。无论是学生报名 教育部门的数据管理，还是其他与学校相关的事务，学校 都扮演着至关重要的角色。那么，如何查询小学的学校 呢？本文将为您提供全面而详细的解答。学籍管理系统查询 学籍管理系统是教育部门为了更好地管理学生信息而建立的一个平台。通过这...
乐华电视质量如何？全面解析乐华电视的画质 音质和耐久性！
乐华电视质量怎么样？乐华电视以其出色的质量而闻名。其电视机采用先进的技术和工艺制造，确保产品的稳定性和可靠性。乐华电视的画质清晰，色彩鲜艳，让观众享受到更逼真的视觉体验。声音方面，乐华电视采用高品质的音响系统，提供清晰 逼真的音效。此外，乐华电视还注重用户体验，提供简洁易用的操作界面和智能功能，让用...