157｜消息丢失防护：生产、存储、消费三个阶段的确认与补偿

生产者应用
  → 序列化（Key/Value Serializer）
  → 分区选择（Partitioner）
  → 消息累加器（RecordAccumulator / 缓冲区）
  → 网络发送（Sender 线程）
  → Broker 接收
  → Broker 返回 ACK

Properties props = new Properties();
props.put("acks", "all");
props.put("retries", Integer.MAX_VALUE);  // 无限重试（Kafka ≥ 2.0 推荐）
props.put("retry.backoff.ms", 100);       // 重试间隔 100ms
props.put("max.in.flight.requests.per.connection", 5);  // 注意顺序性问题

props.put("enable.idempotence", true);

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

try {
    RecordMetadata metadata = producer.send(record).get(5000, TimeUnit.MILLISECONDS);
    log.info("消息发送成功: topic={}, partition={}, offset={}", 
             metadata.topic(), metadata.partition(), metadata.offset());
} catch (TimeoutException e) {
    // 网络超时：重试或写入本地死信表
    saveToLocalDeadLetterTable(record);
} catch (Exception e) {
    // 其他异常：记录并告警
    log.error("消息发送失败", e);
    alertToOncall();
}

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        // 发送失败：降级方案
        log.error("消息发送失败: key={}", record.key(), exception);
        fallbackToDatabase(record);  // 降级写入数据库
    } else {
        log.debug("消息发送成功: offset={}", metadata.offset());
    }
});

public class ReliableMessageSender {
    private final KafkaProducer<String, String> producer;
    private final DeadLetterDao deadLetterDao;
    
    public void sendWithFallback(String topic, String key, String message) {
        ProducerRecord<String, String> record = 
            new ProducerRecord<>(topic, key, message);
        
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                // 1. 写入本地死信表（同一事务保证原子性）
                deadLetterDao.insert(topic, key, message, exception.getMessage());
                // 2. 触发告警
                metrics.counter("mq.send.failure").increment();
                // 3. 定时任务扫描死信表，重试发送
            }
        });
    }
}

// 同步发送——等待 Broker 确认（推荐用于关键业务）
SendResult result = producer.send(msg);
// 返回 SendStatus: SEND_OK / FLUSH_DISK_TIMEOUT / FLUSH_SLAVE_TIMEOUT / SLAVE_NOT_AVAILABLE

// 带重试的同步发送
producer.setRetryTimesWhenSendFailed(3);          // 同步发送失败重试 3 次
producer.setRetryTimesWhenSendAsyncFailed(3);     // 异步发送失败重试 3 次

// 开启 Publisher Confirm
channel.confirmSelect();

// 方式一：逐条确认（性能差）
channel.basicPublish(exchange, routingKey, null, body);
if (channel.waitForConfirms(5000)) {
    // 发送成功
} else {
    // 发送失败，重试
}

// 方式二：批量确认（推荐）
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        // 成功确认
    }
    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        // 失败：重试或告警
    }
});

Producer → Broker 接收
  → 写入 Page Cache（操作系统页缓存）
  → 刷盘（fsync / flush）到磁盘
  → 复制到副本（Replica / Slave）
  → 返回 ACK 给 Producer

Topic: orders (3 partitions, 3 replicas)
Partition 0: Leader=Broker1, ISR=[Broker1, Broker2, Broker3]
Partition 1: Leader=Broker2, ISR=[Broker2, Broker3, Broker1]
Partition 2: Leader=Broker3, ISR=[Broker3, Broker1, Broker2]

# 每个 Topic 的副本数
default.replication.factor=3

# 最小 ISR 数量（acks=all 时，必须至少这么多副本确认才返回成功）
min.insync.replicas=2

# 落后多少毫秒被踢出 ISR
replica.lag.time.max.ms=30000

# 刷盘策略（通常不修改，交给 OS 管理）
log.flush.interval.messages=10000   # 每 10000 条消息刷一次盘
log.flush.interval.ms=1000          # 每 1000ms 刷一次盘

# 默认 false：不允许非 ISR 副本成为 Leader
unclean.leader.election.enable=false

// Broker 配置
flushDiskType=SYNC_FLUSH    // 同步刷盘（默认 ASYNC_FLUSH）
brokerRole=SYNC_MASTER      // 同步复制（默认 ASYNC_MASTER）

// 声明持久化队列
boolean durable = true;
channel.queueDeclare("orders", durable, false, false, null);

// 发送持久化消息
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2)  // 2 = 持久化
    .build();
channel.basicPublish("", "orders", props, body);

// 声明仲裁队列（RabbitMQ 3.8+）
Map<String, Object> args = new HashMap<>();
args.put("x-queue-type", "quorum");
channel.queueDeclare("orders", true, false, false, args);

Broker → 消费者拉取消息
  → 消费者处理业务逻辑（写数据库、调用 RPC...）
  → 消费者提交 Offset（告知 Broker 已消费）

// ❌ 错误模式：自动提交
props.put("enable.auto.commit", "true");  // 危险！
props.put("auto.commit.interval.ms", "1000");

// ✅ 正确模式：手动提交，先处理后提交
props.put("enable.auto.commit", "false");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        try {
            // 1. 处理业务逻辑
            processBusiness(record);
            // 2. 处理成功后提交 Offset
            consumer.commitSync(Collections.singletonMap(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            ));
        } catch (Exception e) {
            log.error("消息处理失败: offset={}", record.offset(), e);
            // 3. 不提交 Offset，消息会被重新消费
            // 或者发送到死信队列
            sendToDeadLetterQueue(record);
            // 仍然提交 Offset 避免阻塞
            consumer.commitSync();
        }
    }
}

// 批量提交（推荐，性能好）
consumer.poll(...);
records.forEach(r -> processBusiness(r));
consumer.commitSync();  // 批量提交当前批次

// 分区级别提交（更精细）
for (TopicPartition partition : records.partitions()) {
    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
    consumer.commitSync(Collections.singletonMap(
        partition, new OffsetAndMetadata(lastOffset + 1)
    ));
}

public class RetryableConsumer {
    private static final int MAX_RETRIES = 3;
    
    public void consume(ConsumerRecord<String, String> record) {
        int retries = 0;
        while (retries <= MAX_RETRIES) {
            try {
                processBusiness(record);
                consumer.commitSync();
                return;  // 成功，退出
            } catch (RetryableException e) {
                retries++;
                if (retries > MAX_RETRIES) {
                    // 超过重试次数，发送到死信队列
                    sendToDeadLetterTopic(record);
                    consumer.commitSync();  // 死信队列写入成功，提交 Offset
                    alertToOncall(record);  // 人工介入
                    return;
                }
                // 指数退避
                Thread.sleep((long) Math.pow(2, retries) * 1000);
            } catch (NonRetryableException e) {
                // 不可重试的异常（如数据格式错误），直接死信
                sendToDeadLetterTopic(record);
                consumer.commitSync();
                return;
            }
        }
    }
}

@Service
public class MessageConsumer {
    
    @Transactional
    public void consume(ConsumerRecord<String, String> record) {
        try {
            // 业务处理
            orderService.createOrder(parseOrder(record.value()));
            // 成功：提交 Offset
            consumer.commitSync();
        } catch (Exception e) {
            // 失败：写入本地消息表（同一事务）
            localMessageDao.insert(
                record.topic(), record.partition(), record.offset(),
                record.value(), "FAILED", 0
            );
            // 仍然提交 Offset，避免阻塞（消息已持久化到本地）
            consumer.commitSync();
        }
    }
}

// 定时补偿任务
@Scheduled(fixedDelay = 60000)
public void retryFailedMessages() {
    List<LocalMessage> failed = localMessageDao.findFailed(MAX_RETRIES);
    for (LocalMessage msg : failed) {
        try {
            orderService.createOrder(parseOrder(msg.getBody()));
            localMessageDao.markSuccess(msg.getId());
        } catch (Exception e) {
            localMessageDao.incrementRetry(msg.getId());
            if (msg.getRetryCount() >= MAX_RETRIES) {
                alertToOncall(msg);
            }
        }
    }
}

consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        try {
            processBusiness(msg);
        } catch (Exception e) {
            // 返回 RECONSUME_LATER，消息会进入重试队列
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

// 关闭自动 ACK
boolean autoAck = false;
channel.basicConsume("orders", autoAck, (consumerTag, delivery) -> {
    try {
        processBusiness(delivery.getBody());
        // 业务处理成功后手动 ACK
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // 拒绝消息，requeue=true 重新入队
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});

┌──────────┐    ┌──────────────────┐    ┌──────────┐
│  业务DB   │    │    消息队列       │    │  消费者   │
│           │    │                  │    │           │
│ 订单表    │───▶│  Topic: orders   │───▶│ 履约系统   │
│ 消息表    │    │  (3副本,acks=all) │    │           │
└──────────┘    └──────────────────┘    └──────────┘
       │                                        │
       │            ┌──────────┐                │
       └───────────▶│  定时对账  │◀───────────────┘
                    │  (补偿)   │
                    └──────────┘

TransactionMQProducer producer = new TransactionMQProducer("tx-producer-group");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 执行本地事务（如创建订单）
            orderService.createOrder((Order) arg);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 回调检查本地事务状态（事务未决时）
        String orderId = msg.getKeys();
        Order order = orderService.getById(orderId);
        if (order != null) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
});

// 发送事务消息
Message msg = new Message("orders", order.toBytes());
producer.sendMessageInTransaction(msg, order);

-- 业务表
CREATE TABLE orders (
    id BIGINT PRIMARY KEY,
    status VARCHAR(20),
    created_at TIMESTAMP
);

-- 本地消息表
CREATE TABLE outbox_message (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    business_id BIGINT NOT NULL,
    business_type VARCHAR(50) NOT NULL,
    topic VARCHAR(100) NOT NULL,
    message_body TEXT NOT NULL,
    status ENUM('PENDING', 'SENT', 'FAILED') DEFAULT 'PENDING',
    retry_count INT DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_status (status, created_at)
);

@Service
public class OrderService {
    
    @Transactional
    public void createOrder(Order order) {
        // 1. 业务操作
        orderDao.insert(order);
        
        // 2. 同一事务写入消息表
        OutboxMessage msg = new OutboxMessage();
        msg.setBusinessId(order.getId());
        msg.setBusinessType("ORDER");
        msg.setTopic("orders");
        msg.setMessageBody(JSON.toJSONString(order));
        outboxMessageDao.insert(msg);
    }
}

// 定时任务：扫描并发送未发送的消息
@Scheduled(fixedDelay = 5000)
public void sendPendingMessages() {
    List<OutboxMessage> pending = outboxMessageDao.findByStatus("PENDING", 100);
    for (OutboxMessage msg : pending) {
        try {
            producer.send(toProducerRecord(msg));
            outboxMessageDao.updateStatus(msg.getId(), "SENT");
        } catch (Exception e) {
            outboxMessageDao.incrementRetry(msg.getId());
        }
    }
}

@Component
public class MessageReconciliation {
    
    @Scheduled(cron = "0 0 2 * * ?")  // 每天凌晨 2 点
    public void reconcile() {
        // 1. 查询业务系统中状态未同步的记录
        List<Order> unsynced = orderDao.findUnsynced(
            LocalDateTime.now().minusDays(1),  // 昨天
            LocalDateTime.now()
        );
        
        for (Order order : unsynced) {
            // 2. 检查下游是否已处理
            if (!fulfillmentService.existsByOrderId(order.getId())) {
                // 3. 补偿：重新发送消息
                producer.send(toProducerRecord(order));
            }
        }
    }
}

订单服务 DB                         消息队列                       履约服务
┌─────────────────┐            ┌──────────────┐            ┌──────────────┐
│  orders 表       │            │ Topic: orders │            │ 消费 + 幂等   │
│  outbox_message  │──定时扫描──▶│ (acks=all,    │──消费者──▶│ 去重表/唯一键 │
│  (同一事务)       │            │  min.isr=2)   │            │              │
└─────────────────┘            └──────────────┘            └──────────────┘