如何使用spring kafka暂停并恢复@KafkaListener

Question

我已经实现了Kafka消费者，现在我有了一个场景。

1. 从Kafka流2.2.5中读取数据。通过Srpingboot发布
2. 加载数据库表1
3. 将数据从表1复制到表2
4. 清理桌子1

要执行上述操作，我需要使用quartz的调度作业（已编写）暂停/恢复Kafka使用者，该作业将数据从表1复制到表2。但是在这个活动中，我希望我的Kafka听众暂停，一旦复制完成，它应该继续。

我的实施：

@KafkaListener(topicPartitions =
    { @TopicPartition(topic = "data_pipe", partitions = { "0" })})
public void listen(ConsumerRecord<String, String> cr) throws Exception {

Answer 1

如果您使用“kafkaListener annotation”自动创建的KafkaListenerEndpointRegistry bean，那么您可以像以下代码一样使用它：

@Component
public class KafkaManager {

    private final KafkaListenerEndpointRegistry registry;

    public KafkaManager(KafkaListenerEndpointRegistry registry) {
        this.registry = registry;
    }
    public void pause() {
        registry.getListenerContainers().forEach(MessageListenerContainer::pause);
    }

    public void resume() {
        registry.getListenerContainers().forEach(MessageListenerContainer::resume);
    }
}

文件：https://docs.spring.io/spring-kafka/reference/html/#pause-resume