spring-cloud-stream-binder-kafka消费者设置手动Offset适配Redis故障场景的方法

1. 基础配置调整 #

首先关闭Kafka消费者自动提交能力，开启手动ACK模式，配置示例如下：

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: 你的Kafka地址:9092
          consumer-properties:
            enable.auto.commit: false # 关闭自动偏移量提交
        bindings:
          你的消费者binding名称-in-0: # 替换为实际的消费者binding名称
            consumer:
              ack-mode: manual # 开启手动确认模式
              auto-offset-reset: earliest # 无偏移量记录时的默认消费位置，按需调整
      bindings:
        你的消费者binding名称-in-0:
          destination: 你消费的Topic名称
          group: 你的消费者组名称

2. 消费逻辑实现 #

消费方法中注入确认对象、消费者对象，在Redis操作成功后再提交偏移量，异常时持久化当前偏移量：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.dao.RedisSystemException;
import org.springframework.data.redis.RedisConnectionFailureException;
import java.util.function.Consumer;

@Component
public class CustomKafkaConsumer implements Consumer<Message<String>> {

    // 自定义的Redis操作服务
    private final RedisService redisService;
    // 自定义的偏移量持久化服务，可存储到本地文件、内嵌数据库等可靠存储
    private final OffsetPersistenceService offsetPersistenceService;

    @Override
    public void accept(Message<String> message) {
        ConsumerRecord<?, ?> record = message.getHeaders().get(KafkaHeaders.CONSUMER_RECORD, ConsumerRecord.class);
        Acknowledgment ack = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
        Consumer<?, ?> consumer = message.getHeaders().get(KafkaHeaders.CONSUMER, Consumer.class);

        String topic = record.topic();
        int partition = record.partition();
        long currentOffset = record.offset();
        String consumerGroup = "你的消费者组名称";

        try {
            // 执行业务逻辑
            handleBusiness(message.getPayload());
            // 执行Redis操作
            redisService.doRedisOperate();
            // 全部操作成功才提交偏移量
            ack.acknowledge();
            // 清理对应偏移量的持久化记录
            offsetPersistenceService.deleteRecord(topic, consumerGroup, partition);
        } catch (RedisConnectionFailureException | RedisSystemException e) {
            // Redis异常场景持久化当前偏移量
            offsetPersistenceService.saveRecord(topic, consumerGroup, partition, currentOffset);
            // 按需选择抛出异常触发重试，或者调用consumer.pause()暂停当前分区消费
            throw new RuntimeException("Redis服务异常，暂停消费等待恢复", e);
        } catch (Exception e) {
            // 其他非Redis业务异常按需处理，确认偏移量或者转入死信队列
            ack.acknowledge();
        }
    }
}

必须保证只有业务逻辑和Redis操作全部执行成功才调用ack方法提交偏移量，异常场景绝对不能提交

3. Redis恢复后的偏移量重置逻辑 #

监听到Redis恢复事件、或者服务启动时，读取持久化的偏移量记录，手动重置消费位置：

import org.apache.kafka.common.TopicPartition;

public void resetConsumeOffset(Consumer<?, ?> consumer, String topic, String consumerGroup) {
    // 读取该消费者组下所有持久化的偏移量记录
    List<OffsetRecord> offsetRecords = offsetPersistenceService.listRecords(topic, consumerGroup);
    for (OffsetRecord record : offsetRecords) {
        TopicPartition topicPartition = new TopicPartition(record.getTopic(), record.getPartition());
        // 重置到异常发生时的偏移量位置，后续从该位置继续消费
        consumer.seek(topicPartition, record.getOffset());
        // 重置完成后删除持久化记录
        offsetPersistenceService.deleteRecord(record.getTopic(), consumerGroup, record.getPartition());
    }
}

注意事项 #

• 偏移量持久化存储需要绑定topic、消费者组、分区三个维度，多实例消费场景下避免出现偏移量覆盖问题
• 可配合定时任务检测Redis可用性，恢复后自动触发偏移量重置和消费恢复，无需人工介入
• 若Redis长时间不可用，可根据业务需求设置重试上限，超过阈值后将消息转入死信队列，避免阻塞正常消费流程

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