@StreamListener已废弃，如何用函数式方式实现Kafka Consumer暂停与恢复？

背景 #

原有的@StreamListener注解已被Spring Cloud Stream废弃，需采用函数式编程模型实现Kafka消费者，并保留基于特定条件暂停/恢复消费者的逻辑。

实现方案 #

通过Spring Cloud Stream的函数式模型，我们可以通过Message对象获取Kafka消费者实例及相关元数据，进而实现暂停/恢复操作。

1. 定义函数式消费者Bean

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.kafka.support.KafkaHeaders;
import java.util.Arrays;
import java.util.Set;
import java.util.function.Consumer;

public class KafkaConsumerConfiguration {

    private final OffsetTracker offsetTracker;
    private Consumer<?, ?> cachedConsumer; // 缓存消费者实例，用于后续恢复操作

    public KafkaConsumerConfiguration(OffsetTracker offsetTracker) {
        this.offsetTracker = offsetTracker;
    }

    @Bean
    public Consumer<Message<PojoClassName>> dataStreamInputConsumer() {
        return message -> {
            // 提取消息体与Kafka元数据
            PojoClassName kafkaInputMsg = message.getPayload();
            String key = message.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, String.class);
            int partitionId = message.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID, Integer.class);
            Consumer<?, ?> consumer = message.getHeaders().get(KafkaHeaders.CONSUMER, Consumer.class);
            Long offset = message.getHeaders().get(KafkaHeaders.OFFSET, Long.class);
            String topic = message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC, String.class);
            Long recordTime = message.getHeaders().get(KafkaHeaders.RECEIVED_TIMESTAMP, Long.class);

            // 缓存消费者实例（每个消费者绑定对应一个实例）
            if (this.cachedConsumer == null) {
                this.cachedConsumer = consumer;
            }

            // 执行业务逻辑
            // ...

            // 按条件暂停消费者分区
            TopicPartition topicPartition = new TopicPartition(topic, partitionId);
            if (offsetTracker.isOffsetExceeding(topicPartition, offset)) {
                Set<TopicPartition> pausedPartitions = consumer.paused();
                if (!pausedPartitions.contains(topicPartition)) {
                    consumer.pause(Arrays.asList(topicPartition));
                    // 可选：记录暂停日志或触发后续通知
                }
            }
        };
    }
}

2. 配置消费者绑定

在application.yml中配置函数式消费者的绑定关系，对应原有的DATASTREAMINPUT通道：

spring:
  cloud:
    stream:
      bindings:
        dataStreamInputConsumer-in-0:
          destination: ${YOUR_TOPIC_NAME} # 替换为实际Topic名称
          group: ${YOUR_CONSUMER_GROUP} # 替换为实际消费组名称
      kafka:
        bindings:
          dataStreamInputConsumer-in-0:
            consumer:
              auto-offset-reset: earliest # 根据业务需求调整
              enable-auto-commit: false # 若需手动管理偏移量，关闭自动提交

3. 实现消费者恢复逻辑

若需基于时间戳或其他条件恢复暂停的分区，可添加定时任务或事件触发逻辑，利用缓存的消费者实例执行恢复：

import org.apache.kafka.clients.consumer.TopicPartition;
import org.springframework.scheduling.annotation.Scheduled;
import java.util.Set;

// 在上述配置类中添加以下方法
@Scheduled(fixedRate = 60000) // 每分钟检查一次，可按需调整频率
public void resumePausedPartitions() {
    if (this.cachedConsumer == null) {
        return;
    }

    Set<TopicPartition> pausedPartitions = this.cachedConsumer.paused();
    if (pausedPartitions.isEmpty()) {
        return;
    }

    // 此处添加你的恢复条件判断（例如：当前时间超过指定时间戳阈值）
    boolean shouldResume = true; // 替换为实际业务逻辑

    if (shouldResume) {
        this.cachedConsumer.resume(pausedPartitions);
        // 可选：记录恢复日志
    }
}

关键说明 #

• 函数式模型中，每个消费者绑定对应一个Consumer<Message<T>>类型的Bean，命名规则为{beanName}-in-0，需与配置文件中的绑定名称一致。
• 通过Message头信息可直接获取Kafka原生Consumer实例，操作方式与原@StreamListener逻辑完全一致。
• 缓存Consumer实例时需注意：分布式部署下，每个实例的消费者独立，暂停/恢复操作仅作用于当前实例分配的分区。

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