Spring Cloud Stream Kafka 按条件触发指定Topic手动消费方案咨询

一、手动调用poll返回空集合的常见原因 #

• 偏移量配置问题：如果消费者配置的auto.offset.reset为默认的latest，新消费者组只会拉取订阅成功后新产生的消息，失败Topic中已有的历史消息不会被拉取；如果该消费者组之前已提交过偏移量到最新位置，也会拉不到历史消息。
• 分区分配未完成：Kafka消费者的subscribe是异步操作，第一次调用poll时可能还未完成分区分配，单次调用10s很可能还没完成分配流程就返回空，通常需要多次调用poll才能拿到首次拉取的消息。
• 配置不匹配：检查Topic名称是否拼写错误、Kafka集群连接地址配置是否正确、消费者是否有该Topic的读取权限，序列化配置是否和消息匹配。
• 自动提交偏移量：如果开启了enable.auto.commit=true，之前测试运行时已经提交了偏移量到最新位置，再次启动消费者时也会拉不到历史消息。

二、适配业务场景的最优实现方案 #

不需要自己手动创建消费者拉取消息，直接使用Spring Cloud Stream Kafka原生提供的能力即可实现需求，同时保证顺序性：

实现步骤 #

1. 控制常规消费的启停
   注入框架提供的BindingsEndpoint，可以直接控制消费绑定的启停，比自己实现消费者更安全稳定：

   // 暂停常规消费
   bindingsEndpoint.changeState("常规消费的binding名称", BindingsEndpoint.State.STOPPED);
   // 恢复常规消费
   bindingsEndpoint.changeState("常规消费的binding名称", BindingsEndpoint.State.STARTED);
   

2. 定义失败Topic的消费绑定
   给失败Topic单独配置一个消费绑定，设置auto-startup: false，项目启动时不会自动启动该消费者：

   spring:
     cloud:
       stream:
         bindings:
           fail-topic-consumer-in-0:
             destination: 你的失败Topic名称
             group: 失败消费组
             consumer:
               auto-startup: false
               concurrency: 1 # 单线程消费保证顺序
   

3. 触发失败消息消费和流程切换
   当监听到数据库恢复正常后，按以下顺序执行：
   • 调用BindingsEndpoint暂停常规消费绑定
   • 启动失败Topic的消费绑定
   • 定时校验失败Topic的消费进度：获取消费组的当前偏移量和对应分区的最大偏移量，当所有分区的消费偏移量都等于分区最大偏移量时，说明所有失败消息已经处理完成
   • 停止失败Topic的消费绑定，恢复常规消费绑定

偏移量校验示例代码 #

@Autowired
private ConsumerFactory<byte[], byte[]> consumerFactory;

public boolean isFailTopicConsumedAll(String topic, String groupId) {
    try (Consumer<byte[], byte[]> consumer = consumerFactory.createConsumer(groupId, "")) {
        List<PartitionInfo> partitions = consumer.partitionsFor(topic);
        List<TopicPartition> topicPartitions = partitions.stream()
                .map(p -> new TopicPartition(topic, p.partition()))
                .toList();
        // 获取各分区最大偏移量
        Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitions);
        // 获取当前消费组的提交偏移量
        for (TopicPartition tp : topicPartitions) {
            OffsetAndMetadata committed = consumer.committed(OffsetFetchSpecification.forTopicPartition(tp));
            if (committed == null || committed.offset() < endOffsets.get(tp)) {
                return false;
            }
        }
        return true;
    }
}

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