Spring Kafka MessageListenerContainer 恢复/暂停 # spring-kafka

Question

由于原生 KafkaConsumer 不是线程安全的，因此不鼓励从不同线程而不是 kafka-consumer 处理线程调用 pause 和 resume 方法。 但由于 spring-kafka 提供了另一层 KafkaMessageListenerContainer，它在内部使用 kafka-consumer。所以我的问题是我们是否可以使用 KafkaListenerEndpointRegistry 通过 id 获取侦听器容器并从其他线程而不是消费者处理线程调用恢复或暂停方法。

kafkaListenerEndpointRegistry.getListenerContainer("id").pause();
       

    ExecutorService executorService  = newFixedThreadPool(2);
    executorService.submit(()->{

        System.out.println("CurrentThread: {}" + Thread.currentThread().getId()+ " " + Thread.currentThread().getName());
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        kafkaListenerEndpointRegistry.getListenerContainer("id").resume();
    });

Answer 1

是的； container.pause() 设置一个标志来告诉Consumer 线程在下一次poll() 调用之前暂停。同样，resume() 重置标志，以便消费者线程将在下一次轮询之前恢复 Consumer。