Apache kafka消费者停止和开始之间丢失的消息

Question

我是一个新的Kafka和使用Apache kafka消费者读取消息从生产者。但当我停下来开始一段时间。之间产生的所有消息都将丢失。如何处理这种情况。我正在使用这些属性“auto.offset.reset”、“latest”和“enable.auto.commit”、“false”。

这是我正在使用的代码。任何帮助都是感激的。

Properties props = new Properties();
        props.put("bootstrap.servers", localhost:9092);
        props.put("group.id", "service");
        props.put("enable.auto.commit", "false"); 
        props.put("auto.offset.reset", "latest"); 
        props.put("key.deserializer", KAFKA_DESERIALIER_STRING_KEYVALUE);
        props.put("value.deserializer", KAFKA_DESERIALIER_STRING_KEYVALUE);

        @SuppressWarnings("resource")
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList(topicname));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {        
                JSONObject jsonObj = new JSONObject(record.value());
                JdbcUtilToUdm.insertdataintodb(args, jsonObj);   
            }
        }   

Answer 1

您必须显式调用consumer.commitSync（）或consumer.commitAsync（），因为您禁用了自动提交。您可以同步或不同步提交，这取决于您需要/喜欢哪种方法。这就是日志中消费者组位置的持久化方式。您应该在处理记录之后调用commit（因此可能是在完成所有插入之后，但在您的情况下是在再次轮询之前）。