腾讯二面：如何保证MQ消息不丢失？重复消费如何保证幂等？

@Service
publicclass ReliableProducer {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Transactional
    public void createOrder(Order order) {
        // 1. 业务数据入库
        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-待发送

        // 事务提交后触发异步发送
        CompletableFuture.runAsync(() -> sendWithRetry(msgId));
    }

    // 这里其实可以使用 xxl-job 等分布式调度框架查询未发送成功的消息发送。
    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 -> {
                    // 更新发送状态
                    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; // 指数退避
        scheduler.schedule(() -> sendWithRetry(msgId), delay, TimeUnit.MILLISECONDS);
    }
}

# Kafka配置示例
acks=all
min.insync.replicas=2 // 最小同步副本数
replication.factor=3  // 每个分区的 总副本数量（含 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. 二次幂等检查（防并发）
            if (isProcessed(msgId)) {
                return;
            }

            // 4. 业务处理
            orderService.processOrder(order);

            // 5. 记录处理状态（设置24小时过期）
            markProcessed(msgId);
        } catch (BusinessException e) {
            // 6. 业务异常处理
            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) {
        // 失败计数
        int failCount = incrementFailCounter(record.key());

        if (failCount < 3) {
            thrownew RetryableException(e); // 触发重试
        } else {
            sendToDlq(record); // 转移死信队列
        }
    }

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

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

        // 消费并发控制
        factory.setConcurrency(3);

        return factory;
    }
}

@Service
@Slf4j
publicclass ReconciliationService {

    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点执行
    public void dailyReconciliation() {
        // 1. 生产端计数
        long produced = countProducerMessages();

        // 2. MQ端计数
        long stored = countMQMessages();

        // 3. 消费端计数
        long consumed = countConsumerMessages();

        // 4. 数据对比
        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("消息丢失告警: 阶段={}, 数量={}", stage, lossCount);
        // 1. 通知运维团队
        alertService.notifyStaff(stage, lossCount);

        // 2. 自动恢复机制
        if (lossCount < 1000) {
            recoveryService.recoverFromBackup();
        } else {
            // 重大事故，启动紧急预案
            emergencyService.handleDisaster();
        }
    }
}