用spring-cloud-stream和函数风格暂停Kafka Consumer

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kafka.KafkaBinderConfiguration;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.messaging.handler.annotation.Payload;

@EnableBinding(Sink.class)
public class KafkaConsumer {

    @Bean
    public KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties() {
        return new KafkaBinderConfigurationProperties();
    }

    @Bean
    public KafkaBinderConfiguration kafkaBinderConfiguration(KafkaBinderConfigurationProperties properties) {
        return new KafkaBinderConfiguration(properties);
    }

    @StreamListener(Sink.INPUT)
    public void process(@Payload String message) {
        // 处理消息的逻辑
        System.out.println("Received message: " + message);
    }

    public void pauseConsumer() {
        // 暂停消费者
        Sink.INPUT.getBinding().pause();
    }

    public void resumeConsumer() {
        // 恢复消费者
        Sink.INPUT.getBinding().resume();
    }
}