19、事件驱动与消息可靠性：消息持久化、消息确认（ACK）、死信队列、重试队列、Kafka与RabbitMQ实践

各位同学，今天我们来聊聊事件驱动架构里最让人头疼，也最绕不开的话题——消息可靠性。

说白了，消息中间件就像快递系统。你寄了个包裹，怎么确保它一定送到？半路丢了怎么办？收件人不在家怎么办？这些问题，在分布式系统里一个不少。

我早年做电商订单系统时，就吃过消息丢失的亏。用户下单成功了，结果积分服务没收到消息，积分没加上。用户投诉过来，我们查了半天日志，才发现是消息在中间件里丢了。从那以后，我对消息可靠性就特别敏感。

19.1 消息持久化：把消息写在磁盘上

消息持久化，就是把消息存到磁盘里。这样即使Broker宕机重启，消息也不会丢。

RabbitMQ的持久化

RabbitMQ的持久化需要做三件事：

  • Exchange设置持久化：durable=true
  • Queue设置持久化：durable=true
  • 消息设置持久化：delivery_mode=2
// RabbitMQ 持久化示例
channel.exchangeDeclare("order.exchange", "direct", true); // durable=true
channel.queueDeclare("order.queue", true, false, false, null); // durable=true

String message = "订单创建成功";
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2) // 持久化消息
    .build();
channel.basicPublish("order.exchange", "order.create", props, message.getBytes());
注意：RabbitMQ的持久化是“写磁盘”操作，性能会下降。我建议只在关键业务消息上开启持久化，日志类消息可以适当放宽。

Kafka的持久化

Kafka天生就是持久化的。所有消息都会写到磁盘的Segment文件中。你不需要额外配置。

Kafka的持久化靠的是日志文件。每个分区对应一个日志目录，消息按顺序追加。这种顺序写磁盘的方式，性能比随机写高很多。

# Kafka 服务端配置，控制日志保留时间
log.retention.hours=168  # 保留7天
log.retention.bytes=1073741824  # 单个日志文件最大1GB

我个人习惯把Kafka的日志保留时间设长一点，比如7天。这样即使消费者挂了几天，重启后还能从头消费。

19.2 消息确认（ACK）：你到底收到没？

消息确认机制，就是消费者告诉Broker：“我收到了，处理完了，你可以删了。” 如果没确认，Broker会重新投递。

RabbitMQ的ACK

RabbitMQ支持手动ACK和自动ACK。自动ACK就是消费者一收到消息就确认，不管处理成功与否。手动ACK是消费者处理完业务逻辑后，再调用basicAck。

// RabbitMQ 手动ACK
channel.basicConsume("order.queue", false, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope,
                               AMQP.BasicProperties properties, byte[] body) throws IOException {
        try {
            // 处理业务逻辑
            processOrder(body);
            // 处理成功，手动ACK
            channel.basicAck(envelope.getDeliveryTag(), false);
        } catch (Exception e) {
            // 处理失败，拒绝消息，重新入队
            channel.basicNack(envelope.getDeliveryTag(), false, true);
        }
    }
});
避坑指南：我曾经在生产环境用过自动ACK，结果消费者处理到一半抛异常了，消息已经确认了，再也收不到了。后来我全部改成手动ACK，处理完业务逻辑再确认。

Kafka的ACK

Kafka的ACK机制在生产者端和消费者端都有。

  • 生产者ACK：控制生产者发送消息后，等待多少个副本确认。
  • 消费者ACK：消费者处理完消息后，提交offset。
// Kafka 生产者ACK配置
Properties props = new Properties();
props.put("acks", "all"); // 等待所有副本确认
props.put("retries", 3);  // 重试次数

// Kafka 消费者手动提交offset
consumer.subscribe(Arrays.asList("order-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(1000);
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    consumer.commitSync(); // 手动提交
}

Kafka的acks=all是最安全的配置，但延迟也最高。我一般用在核心交易链路，非核心链路用acks=1就够了。

19.3 死信队列：处理不了的消息放这里

死信队列，就是消息的“垃圾桶”。当消息处理失败，或者过期了，就会被扔进死信队列。

RabbitMQ的死信队列

RabbitMQ的死信队列需要额外配置。给普通队列绑定一个死信交换机，消息变成死信后，会被路由到死信队列。

// 声明死信交换机
channel.exchangeDeclare("dlx.exchange", "direct", true);
// 声明死信队列
channel.queueDeclare("dlx.queue", true, false, false, null);
channel.queueBind("dlx.queue", "dlx.exchange", "dlx.routingkey");

// 声明普通队列，绑定死信交换机
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx.exchange");
args.put("x-dead-letter-routing-key", "dlx.routingkey");
channel.queueDeclare("order.queue", true, false, false, args);

死信队列有什么用？我举个例子。订单支付超时了，消息变成死信，进入死信队列。然后有个专门的消费者监听死信队列，把超时订单取消掉。这样就把异常处理和正常业务分开了。

Kafka的死信队列

Kafka没有内置的死信队列概念。但我们可以自己实现。比如消费者处理失败后，把消息写入另一个Topic（比如order-dlq），然后单独处理。

// Kafka 手动实现死信队列
try {
    processRecord(record);
} catch (Exception e) {
    // 处理失败，写入死信Topic
    producer.send(new ProducerRecord<>("order-dlq", record.key(), record.value()));
    // 提交offset，避免重复消费
    consumer.commitSync();
}
核心思路：死信队列不是RabbitMQ的专利。在Kafka里，你完全可以自己建一个死信Topic。关键是“分离关注点”——正常业务和异常处理分开。

19.4 重试队列：失败了再试一次

有些失败是暂时的，比如数据库连接超时、网络抖动。这时候重试一下可能就成功了。重试队列就是干这个的。

RabbitMQ的重试队列

RabbitMQ的重试队列通常结合死信队列实现。消息处理失败后，不直接进死信队列，而是先进入一个“重试队列”，等待一段时间后再重新投递。

// 重试队列配置：消息过期后重新投递
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "order.retry.exchange");
args.put("x-message-ttl", 5000); // 5秒后过期
channel.queueDeclare("order.retry.queue", true, false, false, args);

我习惯设置3次重试。第一次失败后等5秒，第二次等10秒，第三次等30秒。如果还失败，就进死信队列人工处理。

Kafka的重试队列

Kafka的重试队列也是自己实现。消费者处理失败后，把消息写入一个“重试Topic”，并设置一个延迟时间。然后另一个消费者监听重试Topic，延迟处理。

// Kafka 重试队列实现思路
// 1. 处理失败，写入重试Topic，带上重试次数
producer.send(new ProducerRecord<>("order-retry", record.key(), record.value()));

// 2. 重试消费者，检查重试次数，决定是否继续重试
if (retryCount < 3) {
    // 延迟5秒后重新处理
    Thread.sleep(5000);
    processRecord(record);
} else {
    // 超过重试次数，进死信队列
    producer.send(new ProducerRecord<>("order-dlq", record.key(), record.value()));
}
注意：重试队列要防止“重试风暴”。如果下游服务挂了，所有消息都在重试，会把系统打垮。我建议加上退避策略，比如指数退避。

19.5 Kafka与RabbitMQ实践对比

这两个中间件我都用过。简单说说我的感受。

特性 RabbitMQ Kafka
消息模型 队列模型，点对点或发布订阅 日志模型，分区顺序消费
持久化 需要显式配置 默认持久化
ACK机制 手动/自动ACK，支持NACK 生产者acks配置，消费者offset提交
死信队列 内置支持，配置简单 需要自己实现
重试队列 结合死信队列和TTL实现 需要自己实现
适用场景 低延迟、复杂路由、事务消息 高吞吐、日志收集、流处理

我个人习惯：业务系统内部的消息用RabbitMQ，因为它的路由灵活，死信队列和重试队列配置方便。大数据和日志场景用Kafka，因为它的吞吐量高，持久化可靠。

19.6 核心知识体系

下面这张图总结了事件驱动与消息可靠性的核心逻辑。你可以看到消息从生产者到消费者，中间经历了持久化、ACK、重试、死信等环节。

事件驱动与消息可靠性核心流程 生产者 消息持久化 Broker 消费者 ACK确认 处理失败 重试队列 死信队列 重试投递 人工处理 正常流程 异常流程 重试流程

从这张图你可以看到，消息从生产者发出，经过持久化存储到Broker，然后投递给消费者。消费者处理成功后发送ACK，Broker删除消息。如果处理失败，消息进入重试队列，重试几次后如果还失败，就进入死信队列，最终人工处理。

嗯，这就是事件驱动与消息可靠性的核心。说白了，就是“确保消息不丢，处理失败有兜底”。

总结：
  • 消息持久化：RabbitMQ显式配置，Kafka默认持久化
  • 消息确认：手动ACK比自动ACK更可靠
  • 死信队列：处理不了的消息放这里，人工介入
  • 重试队列：临时失败先重试，别急着扔垃圾桶
  • Kafka和RabbitMQ各有优势，选型看场景

我在项目中见过太多因为消息丢失导致的数据不一致问题。每次排查起来都特别痛苦。所以，从一开始就把消息可靠性设计好，后面能省很多事。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →