【Java 开发日记】我们来说一下消息的可靠性投递

Ne0inhk

— 9 min read
【Java 开发日记】我们来说一下消息的可靠性投递

目录

1. 核心概念

2. 面临的挑战

3. 关键实现机制

3.1 生产端保证

3.2 Broker端保证

3.3 消费端保证

4. 完整可靠性方案

4.1 事务消息方案（如RocketMQ）

4.2 最大努力投递方案

4.3 本地消息表方案（经典）

5. 高级特性与优化

5.1 顺序性保证

5.2 批量消息可靠性

5.3 监控与对账

6. 不同MQ的实现差异

7. 实践建议

总结

面试回答

1. 核心概念

可靠性投递（Reliable Delivery）是指确保消息从生产者成功到达消费者，即使面对网络故障、系统崩溃等异常情况也能保证不丢失、不重复、按顺序（部分场景）传递。

2. 面临的挑战

  • 网络不可靠：丢包、延迟、分区
  • 节点故障：生产者/消费者/中间件宕机
  • 重复消费：确认机制可能引发重复
  • 顺序保证：分布式环境下消息乱序

3. 关键实现机制

3.1 生产端保证
// 伪代码示例：生产端确认模式 public void sendWithConfirm(Message msg) { // 1. 持久化到本地数据库（防丢失） messageDao.save(msg); // 2. 发送到消息队列 String msgId = rabbitTemplate.convertAndSend(msg); // 3. 等待Broker确认 boolean ack = waitForAck(msgId, TIMEOUT); // 4. 失败重试（指数退避） if (!ack) { retryWithBackoff(msg); } // 5. 最终记录投递状态 updateDeliveryStatus(msgId, ack); }

技术要点：

  • 事务机制：同步方式，性能差（不推荐）
  • 确认机制（Confirm）：
    • 普通确认（每消息确认）
    • 批量确认（提高吞吐）
    • 异步监听（最佳实践）
  • 本地消息表：事务消息的替代方案
  • 消息持久化：设置delivery_mode=2
3.2 Broker端保证
消息处理流程： Producer → Broker接收 → 持久化存储 → 推送给Consumer → 等待ACK → 删除/重投

持久化策略：

  • 队列持久化：durable=true
  • 消息持久化：delivery_mode=2
  • 镜像队列：多副本冗余（RabbitMQ）
  • 高可用集群：主从切换时不丢消息
3.3 消费端保证
// 消费端保证示例 @RabbitListener(queues = "order.queue") public void handleOrder(OrderMessage order, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) { try { // 1. 业务处理 orderService.process(order); // 2. 手动确认（成功才ACK） channel.basicAck(tag, false); // 3. 更新消费记录 consumeRecordService.markConsumed(order.getId()); } catch (Exception e) { // 4. 失败处理：重试或进入死信队列 if (retryCount < MAX_RETRY) { channel.basicNack(tag, false, true); // 重入队列 } else { channel.basicNack(tag, false, false); // 进入死信队列 alarmService.notifyAdmin(order, e); } } }

消费端关键点：

  • 手动ACK：避免自动确认导致消息丢失
  • 幂等性设计：
public boolean processWithIdempotent(String msgId) { // 基于消息ID去重 if (redis.exists("processed:" + msgId)) { return true; // 已处理过 } // 业务处理 boolean success = doBusinessLogic(); // 记录处理状态 if (success) { redis.setex("processed:" + msgId, 24h, "1"); } return success; }
  • 死信队列（DLQ）：处理无法消费的消息
  • 消费重试策略：
    • 立即重试（瞬时故障）
    • 延迟重试（业务依赖）
    • 指数退避（防止雪崩）

4. 完整可靠性方案

4.1 事务消息方案（如RocketMQ）
两阶段提交： 1. 发送Half Message（预备消息） 2. 执行本地事务 3. 根据本地事务结果Commit/Rollback 4. Broker检查事务状态并投递/丢弃
4.2 最大努力投递方案
# 补偿机制实现 def reliable_delivery(message): max_retries = 5 for attempt in range(max_retries): try: # 尝试投递 result = mq_client.send(message) if result.confirmed: log_delivery_success(message.id) return True except Exception as e: log_failure(attempt, e) if attempt == max_retries - 1: # 最终失败，人工介入 send_alert_to_admin(message) save_to_compensation_table(message) return False # 等待后重试 sleep(backoff_time(attempt)) return False
4.3 本地消息表方案（经典）
-- 本地消息表结构 CREATE TABLE local_message ( id BIGINT PRIMARY KEY, biz_id VARCHAR(64), -- 业务ID content TEXT, -- 消息内容 status TINYINT, -- 0:待发送, 1:已发送, 2:已确认 retry_count INT, next_retry_time DATETIME, created_at TIMESTAMP );

工作流程：

  1. 业务数据+消息记录原子性写入本地DB
  2. 定时任务扫描待发送消息
  3. 调用MQ发送，成功后更新状态
  4. 消费者处理完成后反向确认
  5. 对账程序定期校验数据一致性

5. 高级特性与优化

5.1 顺序性保证
  • 全局有序：单队列单消费者（性能低）
  • 局部有序：相同sharding key的消息发到同一队列
  • 牺牲场景：重试队列可能破坏顺序
5.2 批量消息可靠性
// 批量消息的可靠性处理 public class BatchMessageReliableSender { public void sendBatch(List<Message> batch) { // 1. 批量持久化到本地 batchMessageDao.saveAll(batch); // 2. 设置批次ID String batchId = generateBatchId(); // 3. 发送批次消息 boolean success = mqTemplate.sendBatch(batchId, batch); // 4. 批次确认（或单条补偿） if (success) { markBatchDelivered(batchId); } else { // 逐条重试或记录异常 compensateFailedMessages(batch); } } }
5.3 监控与对账
  • 实时监控：
    • 堆积情况监控
    • 消费延迟报警
    • 失败率统计
  • 定期对账：
-- 消息对账SQL示例 SELECT DATE(create_time) as day, COUNT(*) as total_sent, SUM(CASE WHEN status=2 THEN 1 ELSE 0 END) as confirmed, SUM(CASE WHEN status=1 THEN 1 ELSE 0 END) as pending FROM message_record GROUP BY DATE(create_time) HAVING total_sent != confirmed;

6. 不同MQ的实现差异

特性RabbitMQKafkaRocketMQ
可靠性机制确认+持久化+镜像队列副本机制+ACK+Exactly-Once事务消息+本地存储
顺序性单队列保证Partition内有序Queue内有序
事务支持轻量级事务（性能差）支持Exactly-Once语义完整事务消息
最佳适用场景业务消息、高可靠要求日志流、大数据场景金融交易、订单业务

7. 实践建议

  1. 分级可靠性策略：
    • 关键业务：事务消息+本地表+对账
    • 普通业务：确认机制+重试+死信队列
    • 日志类：最多一次投递即可
  2. 性能与可靠性的平衡：
    • 同步刷盘 vs 异步刷盘
    • 同步复制 vs 异步复制
    • 根据业务重要性选择配置
  3. 灾难恢复设计：
# 配置示例：多级降级 mq: primary: url: "amqp://primary" timeout: 1000ms secondary: url: "amqp://secondary" timeout: 2000ms fallback-to-db: true # 最终降级到数据库

总结

消息的可靠性投递是一个系统工程，需要在生产端、Broker端、消费端协同设计，结合业务场景、性能要求、成本约束做出合适的选择。没有"银弹"方案，只有最适合的方案。建议从简单方案开始，随着业务复杂度增加逐步引入更完善的可靠性机制。

面试回答

首先，消息可靠性投递指的是：
一个消息从发送到被消费者成功处理，过程中不会丢失或重复，保证最终数据的一致性。在实际系统里，消息可能因为网络问题、服务重启等原因丢失或重复，所以我们需要一套机制来确保可靠。

为什么需要它呢？
比如在订单系统中，用户支付成功后要通知物流系统，如果消息丢了，物流就不会触发，用户体验就受损；如果消息重复，可能重复发货，造成损失。所以像金融、交易这些场景，可靠性特别重要。

常见的实现方式，我了解的有几种：

  1. 生产者确认机制
    生产者发消息后，MQ（比如RabbitMQ）会返回一个确认（ACK），如果没收到ACK，生产者可以重发。这样可以防止消息在发送阶段丢失。
  2. 消息持久化
    消息保存到磁盘，而不是只放在内存。这样即使MQ重启，消息也不会丢。
  3. 消费者手动ACK
    消费者处理完消息后，手动告诉MQ“我已经处理完了”，MQ才删除消息；如果处理失败，MQ可以把消息重新投递给其他消费者。避免消息在处理阶段丢失。
  4. 事务消息（比如RocketMQ）
    先发一个“半消息”，等本地事务执行成功，再确认投递；如果失败，就回滚。这适用于分布式事务场景。
  5. 消息去重
    为了避免重复消费，可以在消费端做幂等性设计。比如在数据库里记录消息ID，每次处理前先查一下是否已经处理过。

实际中我们一般会结合业务来设计。
比如一个订单状态同步的场景，我可能会用：生产者确认 + 消息持久化 + 消费者手动ACK + 消费端幂等性。这样基本能覆盖发送、存储、消费各个环节的可靠性。

当然，可靠性和性能之间需要权衡，比如持久化会降低吞吐量，手动ACK会增加延迟。所以要根据业务需求来选择合适的方案。

追加：遇到过消息丢失或重复的问题，你是怎么排查和解决的？
追加：是否了解最终一致性、最大努力通知等模式 ？

如果小假的内容对你有帮助，请点赞，评论，收藏。创作不易，大家的支持就是我坚持下去的动力！

Read more

MySQL 进阶：库与表的DDL核心操作全指南（含实战案例）

MySQL 进阶：库与表的DDL核心操作全指南（含实战案例）

🔥草莓熊Lotso：个人主页 ❄️个人专栏: 《C++知识分享》《Linux 入门到实践：零基础也能懂》 ✨生活是默默的坚持，毅力是永久的享受！ 🎬 博主简介： 文章目录 * 前言： * 一. 数据库（库）的核心操作 * 1.1 创建数据库：指定字符集与校验规则 * 1.1.1 语法格式 * 1.1.2 实战案例 * 1.2 字符集与校验规则：影响查询和排序 * 1.2.1 查看系统默认配置 * 1.2.2 查看支持的字符集和校验规则 * 1.2.3 校验规则的实际影响 * 1.3 操纵数据库：查询、修改、

By Ne0inhk
Flutter 组件 powersync_attachments_helper 的适配 鸿蒙Harmony 实战 - 驾驭分布式附件同步、实现鸿蒙端大文件离线存储与生命周期自动化管理方案

Flutter 组件 powersync_attachments_helper 的适配 鸿蒙Harmony 实战 - 驾驭分布式附件同步、实现鸿蒙端大文件离线存储与生命周期自动化管理方案

欢迎加入开源鸿蒙跨平台社区：https://openharmonycrossplatform.ZEEKLOG.net Flutter 组件 powersync_attachments_helper 的适配 鸿蒙Harmony 实战 - 驾驭分布式附件同步、实现鸿蒙端大文件离线存储与生命周期自动化管理方案 前言 在鸿蒙（OpenHarmony）生态的分布式多媒体协作、工业设备故障图片上报以及需要频繁处理大量音频/视频附件的专业级应用开发中，“非结构化数据与 SQL 逻辑的一致性同步”是决定应用能否在大规模复杂场景下存活的技术深水区。面对一条已经同步成功的“设备巡检记录”。如果其关联的“高清故障原图”因为同步时机错位、由于存储空间不足导致的本地缓存被回收，或者是在鸿蒙手机与平板之间由于同步策略不同步导致的文件路径失效。那么不仅会导致用户在查看详情时看到令人沮丧的“附件丢失”占位图，更会严重削弱政务类资产审计的底层严密性。 我们需要一种“逻辑关联、物理对齐”的附件治理艺术。 powersync_attachments_helper 是一套专为 PowerSync 设计的附件同步

By Ne0inhk
Flutter 组件 ansi_text 适配鸿蒙 HarmonyOS 实战：终端色彩渲染，构建高性能 ANSI 日志高亮与命令行交互架构

Flutter 组件 ansi_text 适配鸿蒙 HarmonyOS 实战：终端色彩渲染，构建高性能 ANSI 日志高亮与命令行交互架构

欢迎加入开源鸿蒙跨平台社区：https://openharmonycrossplatform.ZEEKLOG.net Flutter 组件 ansi_text 适配鸿蒙 HarmonyOS 实战：终端色彩渲染，构建高性能 ANSI 日志高亮与命令行交互架构 前言 在鸿蒙（OpenHarmony）生态迈向工业级运维、涉及大量后台守护进程（Daemon）、系统日志审计及开发者工具链（CLI）开发的背景下，如何为枯燥的纯文本终端注入具备视觉层级的色彩与样式，已成为提升调试效率与故障定位速度的“视觉助推器”。在鸿蒙设备这类强调 AOT 极致性能与低级别 shell 交互的环境下，如果应用依然依赖基础的单色字符串输出日志，由于由于信息流极其庞大且缺乏重点，极易由于由于“视觉疲劳”导致关键系统警告或业务异常被淹没在海量数据中。 我们需要一种能够支持 ANSI 转义序列、具备富文本样式（加粗/背景色）且兼容多种终端模拟器的文本渲染方案。 ansi_text 为 Flutter 开发者引入了基于标准

By Ne0inhk
Java 中间件：Dubbo 服务降级（Mock 机制）

Java 中间件：Dubbo 服务降级（Mock 机制）

👋 大家好，欢迎来到我的技术博客！ 📚 在这里，我会分享学习笔记、实战经验与技术思考，力求用简单的方式讲清楚复杂的问题。 🎯 本文将围绕Java中间件这个话题展开，希望能为你带来一些启发或实用的参考。 🌱 无论你是刚入门的新手，还是正在进阶的开发者，希望你都能有所收获！ 文章目录 * Java 中间件：Dubbo 服务降级（Mock 机制） * 什么是服务降级？ * Dubbo Mock 机制简介 * Mock 的触发条件 * Dubbo Mock 的配置方式 * 1. XML 配置方式 * 2. 注解配置方式（推荐） * 3. 自定义 Mock 类 * 4. 强制 Mock（force） * Mock 机制的工作原理 * 实战案例：电商系统中的服务降级 * 场景描述 * 1. 定义服务接口 * 2. 实现

By Ne0inhk