一、先明确核心原则（通用）
消息丢失的风险点主要在 3 个环节：生产端（发送失败）、中间件（宕机 / 重启）、消费端（消费异常未处理），全链路保障的核心是：
- 生产端：确保消息 “必发必达”（发送后得到中间件确认）；
- 中间件：确保消息 “持久化存储”（宕机后不丢失）；
- 消费端：确保消息 “消费完成再确认”（避免消费一半崩溃导致丢失）。
二、RabbitMQ 全链路消息不丢失方案
1. 生产端：确保消息成功发送到 MQ
- 开启生产者确认机制（Publisher Confirm）
核心是让 RabbitMQ 收到消息后，给生产者返回确认回执，生产者只有收到回执才认为发送成功。
- 方式 1：单条确认（同步）：发送一条等一条的确认，性能低，适合少量消息；
- 方式 2：批量确认（异步）：注册回调函数，批量处理确认 / 失败回执，性能高（推荐）。
java// Spring Boot 配置示例 spring:rabbitmq:publisher-confirm-type: correlated # 开启异步确认publisher-returns: true # 开启消息路由失败返回（如队列不存在）java// 生产者代码示例（异步确认） @Autowired private RabbitTemplate rabbitTemplate;@PostConstruct public void init() {// 确认回调：MQ收到消息触发rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {if (!ack) {// 未确认，重试发送（需做幂等，避免重复）log.error("消息发送失败：{}", cause);retrySend(correlationData.getId());}});// 退回回调：消息路由失败触发（如队列不存在）rabbitTemplate.setReturnsCallback(returned -> {log.error("消息路由失败：{}", returned.getMessage());retrySend(returned.getMessage().getMessageProperties().getCorrelationId());}); } - 关键补充：
- 生产端加本地消息表 + 定时重试：若网络抖动导致生产者没收到确认，通过本地消息表（与业务操作同事务）定时重试，保证最终送达；
- 避免使用
mandatory=false（默认）：否则路由失败的消息会直接丢弃，需设为true并通过 ReturnsCallback 处理。
2. RabbitMQ 服务端：确保消息持久化存储
- 队列 + 交换机持久化：创建队列 / 交换机时指定
durable=true，否则 MQ 重启后队列 / 交换机消失，消息丢失；java// 队列声明示例（持久化） @Bean public Queue durableQueue() {return QueueBuilder.durable("durable_queue") // 持久化队列.build(); } - 消息持久化：发送消息时设置
deliveryMode=2（持久化），否则消息只存在内存，MQ 宕机丢失；java// 发送消息时指定持久化 rabbitTemplate.convertAndSend("exchange", "routingKey", "msg", msg -> {msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);return msg; }); - 集群部署：使用镜像队列（Mirror Queue），将队列副本同步到多个节点，避免单节点宕机丢失；
- 配置示例：设置策略
ha-mode=all（所有节点同步）、ha-sync-mode=automatic（自动同步）。
- 配置示例：设置策略
3. 消费端：确保消息消费完成再确认
- 关闭自动确认，使用手动确认（ack）：默认
autoAck=true，消费端刚收到消息就确认，若消费过程崩溃，消息丢失；需设为autoAck=false，消费完成后手动 ack。java// 消费端配置（手动确认） spring:rabbitmq:listener:simple:acknowledge-mode: manual # 手动确认retry:enabled: true # 开启消费重试max-attempts: 3 # 最大重试次数initial-interval: 1000ms # 重试间隔java// 消费端代码示例 @RabbitListener(queues = "durable_queue") public void consume(Message message, Channel channel) throws IOException {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {// 执行业务逻辑processMsg(message);// 消费完成，手动确认（单条确认）channel.basicAck(deliveryTag, false);} catch (Exception e) {log.error("消费失败", e);// 重试后仍失败，拒绝并死信（避免无限重试）channel.basicNack(deliveryTag, false, false);} } - 关键补充：
- 死信队列（DLQ）：消费失败达到重试次数后，将消息转发到死信队列，人工排查，避免消息丢失；
- 消费端幂等：消息可能因重试重复消费，需通过消息 ID / 业务唯一键做幂等处理（如数据库唯一索引）。
三、Kafka 全链路消息不丢失方案
Kafka 的核心设计（分区副本）比 RabbitMQ 更侧重高可用，保障方案略有差异，但核心逻辑一致：
1. 生产端：确保消息成功写入 Kafka
- 开启生产者确认（acks=-1/all）：
acks参数控制生产者收到的确认级别，是生产端防丢失的核心：acks=0：生产者不等待确认，消息可能丢失（禁用）；acks=1：只等 leader 节点确认，leader 宕机且副本未同步则丢失（不推荐）；acks=-1/all：等待 leader + 所有 ISR 副本确认，最高可靠性（推荐）。
properties# 生产者配置 bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092 acks=all retries=3 # 发送失败重试次数 retry.backoff.ms=1000 # 重试间隔 enable.idempotence=true # 开启幂等性，避免重试导致重复消息 - 异步发送 + 回调处理失败：
java
// Kafka生产者代码示例 Producer<String, String> producer = new KafkaProducer<>(props); ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "msg"); producer.send(record, (metadata, exception) -> {if (exception != null) {log.error("发送失败", exception);// 本地消息表重试（同RabbitMQ逻辑）retrySend(record);} });
2. Kafka 服务端：确保消息持久化
- 分区副本机制：创建 topic 时设置
replication.factor≥3（副本数≥3），确保 leader 宕机后，follower 能成为新 leader，消息不丢失；bash# 创建topic（3副本，1分区） kafka-topics.sh --create --topic durable_topic --bootstrap-server kafka1:9092 --replication-factor 3 --partitions 1 - 禁用 unclean.leader.election：设为
false，避免非 ISR 副本成为 leader（可能丢失未同步的消息）； - 消息刷盘策略：默认
log.flush.interval.messages=5000（累计 5000 条刷盘）、log.flush.interval.ms=1000（1 秒刷盘），可根据业务调整（刷盘越频繁，性能越低，可靠性越高）； - 避免数据过期删除：合理设置
retention.ms（消息保留时间，默认 7 天），避免消息未消费就被删除。
3. 消费端：确保消息消费完成再提交 offset
Kafka 通过 offset 记录消费位置，若 offset 提交早于消费完成，会导致消息丢失，核心方案：
- 关闭自动提交 offset，使用手动提交：
properties
# 消费者配置 enable.auto.commit=false # 关闭自动提交 auto.offset.reset=earliest # 消费失败时从最早offset开始（避免漏消息）java// 消费端代码示例（手动提交） KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Collections.singletonList("durable_topic"));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {try {// 执行业务逻辑processMsg(record);} catch (Exception e) {log.error("消费失败", e);// 重试失败则记录，人工处理continue;}}// 批量提交offset（消费完成后提交）consumer.commitSync(); } - 关键补充：
- 消费端幂等：同样需通过消息 key / 业务唯一键做幂等；
- 避免重复消费：使用
consumer.commitSync()（同步提交），而非commitAsync()（异步可能提交失败）； - 死信处理：消费失败的消息可转发到 “死信 topic”，后续人工重试。
四、RabbitMQ vs Kafka 核心差异总结
| 环节 | RabbitMQ 核心保障手段 | Kafka 核心保障手段 |
|---|---|---|
| 生产端 | Publisher Confirm + 本地消息表 | acks=all + 幂等性 + 重试 |
| 服务端 | 队列 / 交换机 / 消息持久化 + 镜像队列 | 副本数≥3 + 禁用 unclean leader 选举 |
| 消费端 | 手动 ack + 死信队列 | 手动提交 offset + 死信 topic |
总结
全链路保证消息不丢失的核心要点：
- 生产端：确认机制 + 重试 + 本地消息表，确保消息必发至中间件；
- 中间件：持久化 + 集群 / 副本，确保消息存储不丢失；
- 消费端：手动确认（ack/offset）+ 死信处理 + 幂等，确保消费完成再标记已处理。
额外建议：
- 核心业务（如交易、支付）优先用 Kafka（副本机制更稳定）或 RabbitMQ 镜像队列；
- 所有环节加监控告警（如生产端重试失败、消费端堆积），及时发现丢失风险。