如何编写脚本/代码来阻止 apache kafka 消费者使用消息？答案

【问题标题】：How do I write a script/code to stop the apache kafka consumer from consuming the messages?如何编写脚本/代码来阻止 apache kafka 消费者使用消息？
【发布时间】：2019-06-07 17:49:31
【问题描述】：

我需要编写一个 Java 代码来启动 kafka 消费者消费消息。一旦消费者从命令行启动。但不确定阻止消费者处理的标准方法。

在我的本地 Windows 机器上，我编写了简单的独立生产者和消费者。现在我是否要阻止独立使用者使用不同的代码/脚本进行进一步处理。 ` private static final String TOPIC = "conftest"; 私有最终静态字符串 BOOTSTRAP_SERVERS = “本地主机：9092”；

private static Consumer<String, String> createConsumer() {
    final Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVERS);
    props.put(ConsumerConfig.GROUP_ID_CONFIG,"test-consumer-group");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
    // Create the consumer using props.
    consumer = new KafkaConsumer<>(props);
    // Subscribe to the topic.
    consumer.subscribe(Collections.singletonList(TOPIC));
    return consumer;
}

public static void main(String args[]) throws Exception {
    runConsumer();
}

private static void runConsumer(){
    Consumer<String, String> consumer = createConsumer();
    try{
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : consumerRecords) {
                System.out.printf("partition = %d, offset = %d, key = %s, value = %s\n", record.partition(), record.offset(), record.key(), record.value()); 
            }
            consumer.commitAsync();
        }
    }catch(WakeupException wue){
        System.out.println("Wake Up Exception Occured");
        wue.printStackTrace();
    }
    catch(Exception e){
        e.printStackTrace();
    }finally {
        consumer.close();
    }
}`

【问题讨论】：

不，不开始。但是要停止正在启动的消费者。
我已经编写了一个独立的 java producer 来启动。我的问题是如何使用不同的 Java 进程来阻止它。
你能把代码贴在这里吗？我不清楚吗？
上面是我用来在 Eclipse 中启动消费者的示例独立代码。
你在寻找close()的方法吗，那只是阻止消费者的方法

标签： java apache-kafka kafka-consumer-api

【解决方案1】：

使用close方法关闭KafkaConsumer只需调用consumer.close();

public void close()

关闭消费者，等待最多 30 秒的默认超时以进行任何需要的清理。如果启用了自动提交，这将在默认超时内尽可能提交当前偏移量。有关详细信息，请参阅 close(long, TimeUnit)。请注意，wakeup() 不能用于中断关闭。

public void close(long timeout, java.util.concurrent.TimeUnit timeUnit) 指定时间

尝试在指定的超时时间内干净地关闭消费者。此方法等待消费者完成挂起的提交并离开组的超时。如果启用了自动提交，这将在超时内尽可能提交当前偏移量。如果消费者无法在超时到期之前完成偏移提交并优雅地离开组，则消费者将被强制关闭。请注意，wakeup() 不能用于中断关闭。

【讨论】：