@Service
publicclass ReliableProducer {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Transactional
    public void createOrder(Order order) {
        // 1. 業(yè)務數(shù)據入庫
        jdbcTemplate.update(
            "INSERT INTO orders(id, amount) VALUES(?, ?)",
            order.getId(), order.getAmount());

        // 2. 消息記錄入庫
        String msgId = UUID.randomUUID().toString();
        jdbcTemplate.update(
            "INSERT INTO message_log(msg_id, topic, message, status) VALUES(?, ?, ?, ?)",
            msgId, "orders", JsonUtil.toJson(order), 0); // 0-待發(fā)送

        // 事務提交后觸發(fā)異步發(fā)送
        CompletableFuture.runAsync(() -> sendWithRetry(msgId));
    }

    // 這里其實可以使用 xxl-job 等分布式調度框架查詢未發(fā)送成功的消息發(fā)送。
    private void sendWithRetry(String msgId) {
        MessageRecord msg = jdbcTemplate.queryForObject(
            "SELECT * FROM message_log WHERE msg_id = ?",
            new MessageRecordRowMapper(), msgId);

        int attempt = 0;
        while (attempt < MAX_RETRIES) {
            try {
                ListenableFuture<SendResult<String, String>> future =
                    kafkaTemplate.send(msg.getTopic(), msg.getMessage());

                future.addCallback(result -> {
                    // 更新發(fā)送狀態(tài)
                    jdbcTemplate.update("UPDATE message_log SET status = 1 WHERE msg_id = ?", msgId);
                }, ex -> {
                    scheduleRetry(msgId, attempt); // 失敗重試
                });

                return;
            } catch (Exception e) {
                scheduleRetry(msgId, attempt);
                attempt++;
            }
        }
    }

    private void scheduleRetry(String msgId, int attempt) {
        long delay = (long) Math.pow(2, attempt) * 1000; // 指數(shù)退避
        scheduler.schedule(() -> sendWithRetry(msgId), delay, TimeUnit.MILLISECONDS);
    }
}

# Kafka配置示例
acks=all
min.insync.replicas=2 // 最小同步副本數(shù)
replication.factor=3  // 每個分區(qū)的 總副本數(shù)量（含 Leader）
unclean.leader.election.enable=false
log.flush.interval.messages=10000
log.flush.interval.ms=1000

@Slf4j
@Component
publicclass ReliableConsumer {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private OrderService orderService;

    @KafkaListener(topics = "orders")
    public void consume(ConsumerRecord<String, String> record) {
        String msgId = record.key();
        Order order = JsonUtil.fromJson(record.value(), Order.class);

        // 1. 冪等檢查
        if (isProcessed(msgId)) {
            log.info("消息重復消費，已跳過: {}", msgId);
            return;
        }

        // 2. 獲取分布式鎖
        Lock lock = redisLockFactory.getLock("LOCK:" + msgId);
        if (!lock.tryLock(3, TimeUnit.SECONDS)) {
            thrownew ConcurrentAccessException("獲取鎖失敗");
        }

        try {
            // 3. 二次冪等檢查（防并發(fā)）
            if (isProcessed(msgId)) {
                return;
            }

            // 4. 業(yè)務處理
            orderService.processOrder(order);

            // 5. 記錄處理狀態(tài)（設置24小時過期）
            markProcessed(msgId);
        } catch (BusinessException e) {
            // 6. 業(yè)務異常處理
            handleFailure(record, e);
        } finally {
            lock.unlock();
        }
    }

    private boolean isProcessed(String msgId) {
        return"PROCESSED".equals(
            redisTemplate.opsForValue().get("MSG:" + msgId));
    }

    private void markProcessed(String msgId) {
        redisTemplate.opsForValue().set(
            "MSG:" + msgId, "PROCESSED", 24, TimeUnit.HOURS);
    }

    private void handleFailure(ConsumerRecord<?, ?> record, Exception e) {
        // 失敗計數(shù)
        int failCount = incrementFailCounter(record.key());

        if (failCount < 3) {
            thrownew RetryableException(e); // 觸發(fā)重試
        } else {
            sendToDlq(record); // 轉移死信隊列
        }
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        // 配置批量ACK（性能與可靠性的平衡）
        factory.getContainerProperties().setAckMode(
            AckMode.BATCH);

        // 消費并發(fā)控制
        factory.setConcurrency(3);

        return factory;
    }
}

@Service
@Slf4j
publicclass ReconciliationService {

    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2點執(zhí)行
    public void dailyReconciliation() {
        // 1. 生產端計數(shù)
        long produced = countProducerMessages();

        // 2. MQ端計數(shù)
        long stored = countMQMessages();

        // 3. 消費端計數(shù)
        long consumed = countConsumerMessages();

        // 4. 數(shù)據對比
        if (produced != stored) {
            handleLoss(produced - stored, "生產到MQ丟失");
        }

        if (stored != consumed) {
            handleLoss(stored - consumed, "MQ到消費丟失");
        }

        log.info("對賬完成: 生產={}, MQ存儲={}, 消費={}",
            produced, stored, consumed);
    }

    private void handleLoss(long lossCount, String stage) {
        log.error("消息丟失告警: 階段={}, 數(shù)量={}", stage, lossCount);
        // 1. 通知運維團隊
        alertService.notifyStaff(stage, lossCount);

        // 2. 自動恢復機制
        if (lossCount < 1000) {
            recoveryService.recoverFromBackup();
        } else {
            // 重大事故，啟動緊急預案
            emergencyService.handleDisaster();
        }
    }
}