不使用DLQ时，如何处理Kafka中因API故障导致的未消费消息

1. 启用手动偏移量提交 #

关闭Kafka Consumer的自动提交配置，仅当外部API调用成功后才提交消费偏移量：

• 核心配置：enable.auto.commit=false
• 代码逻辑示例：

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    try {
        // 调用外部API处理消息
        externalApi.process(record.value());
        // 处理成功后手动提交偏移量
        consumer.commitSync();
    } catch (ApiException e) {
        // API调用失败，不提交偏移量，Consumer会在下一次poll时重新拉取该消息
        log.error("API调用失败，消息将重试", e);
    }
}

注意：这种方式会触发消息重复消费，需确保业务逻辑或API支持幂等。

2. 本地重试+指数退避策略 #

对API调用失败的消息进行本地重试，采用指数退避避免频繁重试占用资源：

int maxRetries = 3;
long initialDelay = 1000; // 初始重试间隔1秒
for (int retry = 0; retry < maxRetries; retry++) {
    try {
        externalApi.process(record.value());
        consumer.commitSync();
        break;
    } catch (ApiException e) {
        if (retry == maxRetries - 1) {
            log.error("达到最大重试次数，消息将进入后续处理流程", e);
            // 此处可触发死信队列或本地持久化逻辑
            break;
        }
        long delay = initialDelay * (long) Math.pow(2, retry);
        Thread.sleep(delay);
        log.warn("API调用失败，第{}次重试", retry + 1);
    }
}

3. 死信队列（DLQ）机制 #

当本地重试失败后，将消息转发到专门的死信Topic，避免阻塞主Topic的消费：

• 创建独立的死信Topic（如main_topic_dlq）
• 失败时发送消息到DLQ：

// 初始化死信Producer
Producer<String, String> dlqProducer = new KafkaProducer<>(dlqProducerConfigs);
// API调用失败且重试耗尽时
dlqProducer.send(new ProducerRecord<>("main_topic_dlq", record.key(), record.value()));
// 提交原Topic的偏移量，继续消费下一条消息
consumer.commitSync();

后续可通过单独的DLQ消费服务，或在原Consumer中添加定时任务，待API恢复后处理死信队列中的消息。

4. 本地持久化失败消息 #

将重试失败的消息存储到本地数据库、Redis或文件系统，通过后台线程定时重试：

• 示例：用Redis List存储失败消息

// API调用失败且重试耗尽时
redisTemplate.opsForList().rightPush("failed_messages", record.value());
// 提交原Topic偏移量
consumer.commitSync();

• 启动后台定时任务：

@Scheduled(fixedRate = 60000) // 每分钟重试一次
public void retryFailedMessages() {
    while (true) {
        String message = redisTemplate.opsForList().leftPop("failed_messages");
        if (message == null) break;
        try {
            externalApi.process(message);
        } catch (ApiException e) {
            // 重试失败，重新放回队列末尾
            redisTemplate.opsForList().rightPush("failed_messages", message);
            break; // 避免长时间阻塞，下次定时任务再尝试
        }
    }
}

关键注意事项 #

• 幂等性保障：无论采用哪种方案，都要确保业务逻辑或外部API支持幂等，避免重复处理消息导致数据异常。
• 监控告警：监控API调用失败率、死信队列消息量、本地持久化失败消息数，及时发现并处理API故障。

内容的提问来源于stack exchange，提问作者Aayush