【Spring Cloud Alibaba】RocketMQ（三）

【Spring Cloud Alibaba】RocketMQ三1. MQ如何保证消息不丢失1.1 哪些环节可能会丢消息1.2 ⽣产者发送消息如何保证不丢失1.2.1 ⽣产者发送消息确认机制1.2.2. Broker 端保证消息持久化不丢1.2.3 消费者端保证消息被真正处理2. MQ如何保证消息的顺序性3. RocketMQ如何保证消息幂等3.1 四种常用幂等方案3.1.1 方案一唯一消息 ID 数据库唯一索引最常用3.1.2 方案二Redis 去重高性能场景3.1.3 方案三状态机幂等订单 / 流程类业务3.1.4 方案四乐观锁 / 版本号3.1.5 生产者端需要做什么4. RocketMQ 如何处理积压消息5. RocketMQ如何保证消息不丢失5.1 生产端确保消息真的到了 Broker5.2 Broker 端确保消息落盘且有副本5.3 消费端确保处理成功才确认6. RocketMQ 接收到的消息存在内存中还是磁盘中 如果消息积压那么消息存在哪里1. MQ如何保证消息不丢失1.1 哪些环节可能会丢消息⾸先分析下MQ的整个消息链路中有哪些步骤是可能会丢消息的。其中1、2、4 三个场景都是跨⽹络的⽽跨⽹络就肯定会有丢消息的可能。然后关于3这个环节通常MQ存盘时都会先写⼊操作系统的缓存page cache中然后再由操作系统异步的将消息写⼊硬盘。这个中间有个时间差就可能会造成消息丢失。如果服务挂了缓存中还没有来得及写⼊硬盘的消息就会丢失。1.2 ⽣产者发送消息如何保证不丢失⽣产者发送消息之所以可能会丢消息都是因为⽹络。因为⽹络的不稳定性容易造成请求丢失。怎么解决这样的问题呢其实⼀个统⼀的思路就是⽣产者确认。简单来说就是⽣产者发出消息后给⽣产者⼀个确定的通知这个消息在Broker端是否写⼊完成了。就好⽐打电话不确定电话通没通那就互相说个“喂”具体确认⼀下。只不过基于这个同样的思路各个MQ产品有不同的实现⽅式。1.2.1 ⽣产者发送消息确认机制在RocketMQ中提供了三种不同的发送消息的⽅式/** * RocketMQ 原生Producer三种发送方式 * 1. oneway单向发送只管发送不等待broker响应性能最高存在消息丢失风险 * 2. sync同步发送阻塞等待broker返回确认结果消息可靠性最高吞吐量较低 * 3. async异步发送不阻塞业务线程通过回调接收结果兼顾性能与可靠性消耗客户端线程资源 */// 1、单向发送 OneWay// 特点发送之后直接返回不需要 Broker 返回确认性能极高但有丢消息风险// 使用场景日志收集、统计埋点允许少量消息丢失的场景producer.sendOneway(msg);// 2、同步发送 Sync// 特点业务线程阻塞等待 Broker 返回发送结果消息可靠性最高吞吐效率偏低// send第二个参数为超时时间单位毫秒这里设置20秒超时// 使用场景重要业务消息订单、支付要求确认消息已经投递成功SendResultsendResultproducer.send(msg,20*1000);// 3、异步发送 Async// 特点业务线程不会阻塞内部新开线程等待Broker响应通过SendCallback回调接收成功/失败结果// 权衡兼顾可靠性与发送效率但会占用客户端额外线程大量消息时注意线程池压力// 使用场景接口不能长时间阻塞同时又要保证消息不丢失的业务场景producer.send(msg,newSendCallback(){/** * 消息发送成功回调 * param sendResult 发送结果包含msgId、队列信息等 */OverridepublicvoidonSuccess(SendResultsendResult){// 发送成功后的业务处理例如记录msgId日志}/** * 消息发送异常回调 * param e 发送失败异常网络、broker异常、超时等 */OverridepublicvoidonException(Throwablee){// 发送失败处理可做重试、告警、落盘记录}});禁用sendOneway单向发送—— 它不等待 Broker 应答只适合日志埋点。业务消息用同步发送send()或异步发送 SendCallback回调。开启发送失败重试设置retryTimesWhenSendFailed同步和retryTimesWhenSendAsyncFailed异步。异步必须处理onException失败时做重试或本地落盘告警不能静默吞掉。校验SendResult状态只有SendStatus.SEND_OK才算发送成功。强一致场景用事务消息或本地消息表保证 “业务数据库操作” 和 “发消息” 的原子性避免库成功但消息没发出去。1.2.2. Broker 端保证消息持久化不丢同步刷盘SYNC_FLUSH消息真正写入磁盘后才返回成功默认的异步刷盘ASYNC_FLUSH只写 PageCache机器断电会丢消息。同步主从复制SYNC_MASTERMaster 写完后等待 Slave 同步完成再返回默认异步复制下 Master 宕机会丢失尚未同步的消息。多副本集群部署避免单点故障。高可靠组合 同步刷盘 同步主从复制代价是 TPS 下降。1.2.3 消费者端保证消息被真正处理业务执行成功后才提交 offsetRocketMQ 默认在onMessage正常返回后才提交消费位点抛出异常则不提交消息会重试投递。禁止吞异常catch 住异常只打日志不抛出框架会误以为消费成功offset 直接提交消息就丢了。正确做法是打印日志后继续向上抛出异常。监控死信队列DLQ超过最大重试次数maxReconsumeTimes的消息会进入死信队列不监控等同于消息丢失。业务幂等RocketMQ 是At-Least-Once至少一次投递重试会导致重复消费业务侧必须做幂等。2. MQ如何保证消息的顺序性同一组需要有序的消息发送到同一个 MessageQueue消费者对这个 MessageQueue 串行消费。https://blog.csdn.net/Blue_Pepsi_Cola/article/details/163996069?spm1001.2014.3001.55013. RocketMQ如何保证消息幂等https://rocketmq.apache.org/zh/docs/bestPractice/01bestpracticeRocketMQ 的投递语义是At-Least-Once至少一次消息可能被重复投递但RocketMQ 本身不提供 Exactly-Once恰好一次语义。幂等性必须由业务消费端自己保证。重复消费的来源消费者处理完消息但提交 offset 前宕机、网络超时导致 Broker 认为消费失败、消费异常触发重试等。幂等的核心思路利用消息的唯一标识在执行业务前判断 “这条消息是否已经处理过”处理过则直接跳过。3.1 四种常用幂等方案3.1.1 方案一唯一消息 ID 数据库唯一索引最常用发送消息时携带全局唯一 ID如订单号、业务流水号消费端用这个 ID 做去重。OverridepublicvoidonMessage(MessageExtmsg){// 1. 从消息中取出唯一业务ID推荐用业务自身的唯一键如订单号StringorderNomsg.getKeys();// 生产者发送时 setKeys(订单号)// 2. 插入去重表利用唯一索引防重// 插入成功 → 第一次消费执行业务// 插入失败主键冲突 → 重复消费直接跳过try{dedupMapper.insert(orderNo);// INSERT INTO dedup_log(biz_id) VALUES(?)}catch(DuplicateKeyExceptione){log.warn(重复消息跳过消费: {},orderNo);return;// 幂等返回不执行业务}// 3. 执行业务逻辑doBusiness(orderNo);}CREATETABLEmsg_dedup_log(idBIGINTPRIMARYKEYAUTO_INCREMENT,biz_idVARCHAR(64)NOTNULLCOMMENT业务唯一ID,create_timeDATETIMEDEFAULTCURRENT_TIMESTAMP,UNIQUEKEYuk_biz_id(biz_id))COMMENT消息消费去重表;3.1.2 方案二Redis 去重高性能场景用 Redis 的SETNX不存在才设置做幂等判断适合高并发、对性能要求高的场景。OverridepublicvoidonMessage(MessageExtmsg){StringbizIdmsg.getKeys();StringdedupKeyrocketmq:dedup:bizId;// SETNXkey不存在才设置成功返回1已存在返回0BooleanisFirstredisTemplate.opsForValue().setIfAbsent(dedupKey,1,24,TimeUnit.HOURS);// 设置过期时间避免key无限增长if(Boolean.FALSE.equals(isFirst)){log.warn(重复消息跳过: {},bizId);return;}// 执行业务doBusiness(bizId);}优点性能极高缺点Redis 宕机或 key 过期可能导致幂等失效不能用于强一致业务。建议 Redis DB 双保险。3.1.3 方案三状态机幂等订单 / 流程类业务对于有状态流转的业务如订单待支付 → 已支付 → 已发货通过状态判断天然实现幂等。OverridepublicvoidonMessage(MessageExtmsg){StringorderNomsg.getKeys();OrderorderorderMapper.selectByOrderNo(orderNo);// 状态判断只有待支付状态才能更新为已支付// 已经是已支付或更后状态说明重复消息直接返回if(order.getStatus()!OrderStatus.PENDING_PAY){log.warn(订单状态不匹配跳过: current{},order.getStatus());return;}// 带条件更新影响行数0说明被并发消费了也是幂等introwsorderMapper.updateStatusToPaid(orderNo);if(rows0){log.warn(并发更新跳过);return;}// 后续业务doBusiness(order);}UPDATEt_orderSETstatusPAIDWHEREorder_no#{orderNo}ANDstatusPENDING_PAY;这是最优雅的幂等方案不需要额外去重表利用业务自身状态约束。3.1.4 方案四乐观锁 / 版本号在业务表加version字段更新时携带版本号版本不匹配则更新失败。UPDATEt_accountSETbalancebalance-#{amount},versionversion1WHEREaccount_id#{accountId}ANDversion#{version};更新影响行数为 0 → 重复消息或并发冲突 → 跳过。3.1.5 生产者端需要做什么幂等虽然主要在消费端但生产者必须提供唯一标识// 发送消息时设置 keys作为幂等判断依据MessageStringmessageMessageBuilder.withPayload(msgBody).setHeader(RocketMQHeaders.KEYS,orderNo)// 业务唯一ID.build();rocketMQTemplate.syncSend(topic,message);推荐用业务自身的唯一键订单号、流水号而不是 RocketMQ 的msgId—— 因为重试时 msgId 可能变化业务键才是稳定的。RocketMQ 是至少一次投递会重复消费幂等由业务端保证。核心思路是用消息的唯一业务 ID 做去重常用方案有四种 —— 数据库唯一索引去重表最通用、Redis SETNX高性能、状态机判断有状态业务首选、乐观锁版本号并发更新。生产者发送时要把业务唯一 ID 放进 message keys消费者执行业务前先判断是否已处理处理过则直接返回。4. RocketMQ 如何处理积压消息消息积压的问题就是生产者生产消息的速度大于消费者消费消息的速度。通过RocketMQ控制台通过延迟能看到有多少积压的消息也需要我们分场景在不同的场景下使用不同的方法来解决消息积压。①如果是洪峰流量但是消息又不是那种要求瞬时消费完成的就慢慢等待消费完成即可。清理时间积压消息数 ÷消费TPS-生产TPS 如果消费TPS小于或等于生产TPS积压就不会下降需要提升消费能力或者限制生产速度。②另一种问题是消费者因为自身性能的原因所导致的消费能力的低下比如存在慢SQL或者存在性能逻辑的代码第三方调用等等的代码这种就需要优化慢SQL还有调用耗时的代码。解决方案① 横向扩容消费者如果对实时性要求比较高的话像下图一个topic下有多个MessageQueue但是只有一个Consumer对于这种我们可以利用消费者组上线多个消费者Consumer来增加消费者组的消费能力甚至可以一个MessageQueue对应一个消费者来进行消费。Topic├──MessageQueue0──Consumer1├──MessageQueue1──Consumer2├──MessageQueue2──Consumer3└──MessageQueue3──Consumer4注意“一个 MessageQueue 对应一个 Consumer”不是手动绑定而是由 ConsumerGroup 自动负载均衡在 RocketMQ 4.x 队列级负载均衡下消费者实例数超过 MessageQueue 数量后继续扩容通常没有效果。https://rocketmq.apache.org/zh/docs/domainModel/08consumergroup解决方案② 提高单实例消费并发除了增加消费者实例还可以适当增加单个消费者的消费线程数。RocketMQ 4.x Java 客户端示例DefaultMQPushConsumerconsumernewDefaultMQPushConsumer(order_consumer_group);consumer.setConsumeThreadMin(40);consumer.setConsumeThreadMax(80);线程数不能无限增加应根据以下资源确定上限数据库连接池容量下游接口限流阈值CPU 和内存单条消息平均耗时消费成功率和超时率如果增加线程后消费 TPS 没有提高反而出现更多超时说明瓶颈已经转移到了数据库或下游服务。解决方案③ 新增topic在新的topic下跑RocketMQ 消息积压的本质是生产速度大于消费速度。遇到积压时我不会直接扩容而是先通过 Dashboard 查看积压量、消费延迟、生产 TPS、消费 TPS、失败率以及重试队列判断是突发流量还是消费者异常。如果只是短时间流量洪峰并且业务对实时性要求不高可以让消费者慢慢消化。预计清理时间可以用下面的公式计算清理时间积压数量 ÷消费TPS-生产TPS如果是消费者性能问题就需要排查慢 SQL、数据库连接池、第三方接口超时、锁竞争、消费线程阻塞、频繁重试以及 JVM GC 等问题。可以通过批量查询、批量写入、合并 RPC、减少同步日志、缩小事务范围等方式优化单条消息的处理耗时。如果业务实时性要求比较高可以增加同一个 ConsumerGroup 下的消费者实例或者提高单实例的消费线程数从而提升消费并发。但在 RocketMQ 4.x 的队列级负载均衡模式下有效消费者实例数通常不能超过 Topic 的 MessageQueue 数量超过后多出的实例可能处于空闲状态。MessageQueue 和 Consumer 的分配由 RocketMQ 自动完成不需要手动绑定。如果历史积压非常严重又必须保障新消息的实时性还可以采用新旧 Topic 隔离方案历史消息old_topic → old_consumer_group → 限速清理 新消息new_topic → new_consumer_group → 实时消费生产者切换到新 Topic新消费者负责处理实时消息旧消费者继续消费旧 Topic直到积压清空。这个方案需要注意消息顺序因为新消息可能比旧消息先执行所以消费端必须支持幂等、状态校验和事件版本控制。同时新旧消费者如果访问同一个数据库还要进行限流防止把下游服务打垮。对于反复消费失败的毒消息不应该在消费线程中无限重试而应该设置合理的重试次数超过次数后进入死信队列再通过补偿任务单独处理。如果积压消息已经失效并且业务明确允许丢弃也可以通过重置消费位点跳过历史消息但这是有数据丢失风险的操作必须经过业务确认。最后为了避免问题再次发生还需要建立积压量、消费延迟、消费失败率、重试量和死信消息的监控告警并保证消费逻辑幂等。5. RocketMQ如何保证消息不丢失消息丢失只可能发生在三个环节所以要分三段来保证生产者发送、Broker 存储、消费者消费。5.1 生产端确保消息真的到了 Broker用同步发送并检查结果。最常见的丢失原因就是用了 sendOneway 或者异步发送却不处理回调SendResultresultproducer.send(msg);if(result.getSendStatus()!SendStatus.SEND_OK){// 落库重试或记录补偿日志}SendStatus 有四种只有 SEND_OK 是真正安全的。FLUSH_DISK_TIMEOUT、FLUSH_SLAVE_TIMEOUT、SLAVE_NOT_AVAILABLE 都意味着消息可能只在内存里或只在主节点上。配置重试次数producer.setRetryTimesWhenSendFailed(3),同步发送默认重试 2 次且会切换到其他 Broker。极端场景用事务消息。如果要求「本地数据库事务成功 ⟺ 消息一定发出」重试也不够,进程在「写库成功、发消息前」崩溃就丢了。这时需要事务消息的半消息 回查机制或者自己做本地消息表 定时扫描补偿。5.2 Broker 端确保消息落盘且有副本这是最容易被忽略的一环,默认配置是会丢消息的。刷盘策略。默认flushDiskTypeASYNC_FLUSH消息先进PageCache就返回成功机器断电直接丢失。要绝对不丢得改同步刷盘flushDiskType SYNC_FLUSH代价是吞吐量下降明显一个数量级所以实践中多数场景选异步刷盘 主从复制来折中。主从复制策略。单节点再怎么刷盘磁盘坏了还是丢。需要brokerRole SYNC_MASTER # 同步复制等从节点确认后才返回成功ASYNC_MASTER 下主节点挂掉未同步的消息就没了。更彻底的方案是 DLedgerRaft模式多副本自动选主写入要多数派确认。5.x 推荐这个它同时解决了丢消息和主节点故障后不可写的问题传统主从里从节点不能自动升主。5.3 消费端确保处理成功才确认核心原则业务处理完成后再返回成功状态。consumer.registerMessageListener((MessageListenerConcurrently)(msgs,context)-{try{handle(msgs);returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;}catch(Exceptione){// 返回 RECONSUME_LATER消息会进重试队列returnConsumeConcurrentlyStatus.RECONSUME_LATER;}});两个典型错误写法一是用异步线程池处理消息后立刻返回 CONSUME_SUCCESS,offset 提交了但业务还没做完二是 catch 住所有异常后仍然返回成功异常被吞掉。重试与死信队列。返回 RECONSUME_LATER 后 RocketMQ 会按延迟级别重试默认 16 次后进入死信队列%DLQ%{consumerGroup}。死信队列必须有人监控和处理否则消息相当于静默丢失了。消费幂等是必须的。上面所有机制保证的都是「至少一次」投递不是「恰好一次」。生产者重试、Broker 主从切换、消费重试都会造成重复所以业务侧要用业务唯一键做去重数据库唯一索引或 Redis 判重。6. RocketMQ 接收到的消息存在内存中还是磁盘中 如果消息积压那么消息存在哪里RocketMQ 的消息始终写在磁盘上但写入和读取都经过操作系统的 PageCache。所以准确说法是消息持久化在磁盘热数据缓存在内存PageCache而不是存在 JVM堆内存里。这一点和 RabbitMQ 那种「内存队列 内存不够才换页到磁盘」的设计完全不同——RocketMQ 没有「内存模式」所有消息一律落 CommitLog 文件。写入路径Broker 用 mmap 把 CommitLog 文件映射到虚拟内存写消息就是顺序追加到映射区,这一步只写进了 PageCache操作系统随后负责落盘。刷盘时机由配置决定ASYNC_FLUSH默认写入 PageCache 即返回成功后台线程异步刷盘。断电会丢未刷盘部分SYNC_FLUSH强制 force() 落盘后才返回成功安全但吞吐降一个数量级顺序写 PageCache 是 RocketMQ 磁盘存储却能有高吞吐的根本原因,顺序写磁盘的速度可以接近内存随机写。还有一个可选优化 transientStorePoolEnabletrue消息先写堆外内存DirectByteBuffer池再由 commit 线程批量写入PageCache读写分离降低毛刺代价是异步丢失窗口变大。磁盘上的三类文件消费时的路径是先读 ConsumeQueue 拿到物理偏移20 字节一条极小几乎全在 PageCache 里再去 CommitLog 捞消息体。消息积压时消息在哪里还在磁盘的 CommitLog 文件里位置完全没变。这里要理解 RocketMQ 的一个关键设计消息存储与消费状态是分离的。消息写进 CommitLog 后就不动了,所谓「积压」只是消费组的 consumerOffset 落后于队列的maxOffset是两个数字的差值不是消息堆在某个地方。所以有三个推论积压不会撑爆 Broker 的 JVM 内存。 积压一亿条消息Broker 堆内存占用和不积压时基本一样。这和 RabbitMQ 积压时内存暴涨、触发流控的行为完全不同。积压真正的代价是「冷读」。 正常消费时读的是刚写入的消息还在 PageCache 里命中率接近100%速度极快。一旦积压严重消费者要去读几小时前的数据PageCache早已被新消息挤出去,于是变成磁盘随机读耗时从微秒级跳到毫秒级。这会形成恶性循环消费变慢 → 积压更多 → 读得更冷 →更慢。这是线上处理积压最棘手的地方而不是存储空间。积压最危险的后果是消息被删除。 RocketMQ 删除 CommitLog 只看时间和磁盘水位不看有没有被消费完fileReservedTime 72 # 文件保留 72 小时deleteWhen 04 # 每天凌晨 4 点执行删除diskMaxUsedSpaceRatio 75 # 磁盘使用率超 75% 开始清理磁盘水位的强制动作更激进达到 85%diskSpaceCleanForciblyRatio强制删除最老的文件不管是否过期、是否被消费达到 90%diskSpaceWarningLevelRatioBroker 拒绝写入producer 收到 SERVICE_NOT_AVAILABLE也就是说积压超过 72 小时或者把磁盘写满消息就真的丢了。这是积压场景下最现实的丢消息原因比刷盘策略和主从复制更常见。消费端的内存消费者侧确实有内存缓存但有流控兜底。PushConsumer 的后台线程把消息批量拉到本地 ProcessQueue内存中的TreeMap再投给业务线程池。防止拉爆内存的参数consumer.setPullThresholdForQueue(1000); // 单队列最多缓存 1000 条consumer.setPullThresholdSizeForQueue(100); // 单队列最多缓存 100MBconsumer.setConsumeThreadMax(64);触发阈值后暂停拉取。所以即使 Broker 端积压上亿条消费端内存也是有界的——这正是「拉模式 流控」相比推模式的优势。小结处理积压的实践顺序通常是先确认磁盘水位和 fileReservedTime还剩多少余量防止边处理边丢再扩消费者实例注意不能超过队列数否则扩了也闲置必要时临时把消息转发到新 Topic 用更多队列并行消费。