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

Question

由于本机KafkaConsumer不是线程安全的，因此不鼓励从不同的线程而不是kafka使用者处理线程调用pause和resume方法。但正如spring kafka提供的另一层kafka信息容器，内部使用kafka consumer。所以我的问题是，我们可以使用KafkaListenerEndpointRegistry通过id获取侦听器容器，并从其他线程而不是消费者处理线程调用resume或pause方法吗。

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（）设置一个标志，告诉消费者线程在下一个轮询（）调用之前暂停。类似地，恢复（）重置标志，以便消费者线程将在下一次轮询之前恢复消费者。