Kafka有批量消费者吗？

Question

高级使用者 API 似乎一次读取一条消息。

如果消费者想要处理这些消息并提交给其他下游消费者(如Solr或Elastic-Search ),这可能会给他们带来很大的问题，因为他们更喜欢批量接收消息，而不是一次接收一条。

在内存中批处理这些消息也并非易事，因为只有当批处理已经提交时，Kafka中的偏移量也需要同步，否则具有未提交下游消息的崩溃的 kafka 使用者（如在Solr或ES中）将已经更新其偏移量，因此消息松散。

如果使用者在下游提交消息后但在更新消息偏移量之前崩溃，则该使用者可能会多次使用消息。

如果 Kafka 批量使用消息，那么一些指向代码/文档的指针将不胜感激。

谢谢！

Answer 1

我不知道批量消费者。但即使有一个，你的主要问题仍然存在。您希望在成功转发数据后提交偏移量。实现此目的的一种方法是通过设置属性 auto.commit.enable = false 来关闭使用者的自动提交。当然，权衡是您必须注意何时提交偏移量。

在此处查找使用者属性的完整文档：https://kafka.apache.org/documentation.html#consumerconfigs

关于如何手动提交从java文档中窃取的偏移量的一个很好的示例(https://kafka.apache.org/090/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html):

 Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("group.id", "test");
 props.put("enable.auto.commit", "false");
 props.put("auto.commit.interval.ms", "1000");
 props.put("session.timeout.ms", "30000");
 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
 consumer.subscribe(Arrays.asList("foo", "bar"));
 final int minBatchSize = 200;
 List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
 while (true) {
     ConsumerRecords<String, String> records = consumer.poll(100);
     for (ConsumerRecord<String, String> record : records) {
         buffer.add(record);
     }
     if (buffer.size() >= minBatchSize) {
         insertIntoDb(buffer);
         consumer.commitSync();
         buffer.clear();
     }
 }