kafka的poll

在Kafka中，一个消费者通过调用 poll() 方法来拉取消息。在调用该方法时，消费者将会获得订阅的所有主题分区中已经准备好的消息记录。如果没有任何消息记录可以获取，poll() 方法将会阻塞并等待，直到有消息记录可以被获取为止。在获取到消息记录后，消费者可以对这些消息进行处理，处理完毕后再次调用 poll() 方法来获取更多的消息记录。

需要注意的是，消费者通过 poll() 方法获取的消息记录是按照分区顺序进行返回的。也就是说，如果一个消费者订阅了多个主题分区，那么它通过 poll() 方法获取的消息记录可能会包含多个主题分区中的消息。

相关问题

kafka poll

Kafka poll 是 Kafka 客户端库中的一个方法，用于从 Kafka 服务器中拉取数据记录。当应用程序使用 Kafka Consumer 客户端订阅了一个或多个主题后，它可以使用 poll 方法来主动拉取数据。这个方法会阻塞，直到有数据可用或等待超时。

使用 Kafka poll 方法，应用程序可以周期性地检查 Kafka 服务器上的新数据，然后进行相应的处理。这种拉取的方式相比推送模型更加灵活，使得应用程序可以按需获取数据，并能够控制数据的处理速度。

在调用 poll 方法时，可以指定一个超时时间。如果在指定的时间内没有可用的数据，则 poll 方法将返回一个空的记录集合。通过调整超时时间，可以根据应用程序的需求来平衡实时性和资源消耗。

需要注意的是，Kafka poll 方法是一个阻塞操作，它会阻塞当前线程，直到有数据可用或等待超时。因此，在使用 Kafka Consumer 客户端时，需要小心控制 poll 方法的调用方式，以免阻塞过久导致应用程序的响应性能下降。

kafka poll 过时

从 Kafka 2.0 版本开始，使用 poll 方法消费 Kafka 消息已被标记为过时的 API。相反，建议使用 subscribe 和 assign 方法进行消费。subscribe 方法允许使用组管理（group management）来消费主题上的消息，而 assign 则允许直接分配分区（partitions）来消费消息。这些方法的使用方式与 poll 方法略有不同，需要先创建一个 KafkaConsumer 对象，然后调用相应的方法来订阅或分配分区。以下是一个使用 subscribe 方法进行消费的示例：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
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("test-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

这个示例中，我们首先创建了一个 KafkaConsumer 对象，并使用 subscribe 方法订阅了一个主题。然后在一个无限循环中，调用 poll 方法来获取消息记录，并对每条记录进行处理。需要注意的是，poll 方法的参数指定了等待时间，这里我们设置为 100 毫秒。如果 poll 方法在等待时间内没有获取到任何消息，它会返回一个空的 ConsumerRecords 对象。