KafkaConsumer怎么获取获取是否存在未消费消息

KafkaConsumer 可以通过调用 `poll` 方法获取消息并处理，如果没有消息可处理，`poll` 方法会一直阻塞等待。如果你想检查是否存在未消费的消息，可以通过调用 `endOffsets` 方法获取每个分区的最新偏移量，然后再通过调用 `position` 方法获取当前消费者的偏移量，比较二者是否相等即可。

具体代码如下：

Map<TopicPartition, Long> endOffsets = kafkaConsumer.endOffsets(kafkaConsumer.assignment());
for (TopicPartition partition : endOffsets.keySet()) {
    Long endOffset = endOffsets.get(partition);
    Long currentOffset = kafkaConsumer.position(partition);
    if (endOffset > currentOffset) {
        // 存在未消费的消息
    }
}

以上代码会获取所有消费者订阅的分区的最新偏移量，并逐个比较当前消费者的偏移量与最新偏移量，判断是否存在未消费的消息。

相关问题

kafka java动态获取topic并动态创建消费者

可以使用 Kafka 的 AdminClient API 动态获取 topic 列表，并使用 KafkaConsumer API 动态创建消费者。

首先，使用 AdminClient API 获取所有的 topic 列表：

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

AdminClient adminClient = AdminClient.create(props);
ListTopicsOptions options = new ListTopicsOptions();
options.listInternal(true); // 包括内部 topic
List<String> topicList = adminClient.listTopics(options).names().get();

然后，使用 KafkaConsumer API 动态创建消费者：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 动态订阅 topic
for (String topic : topicList) {
    consumer.subscribe(Collections.singletonList(topic));
}

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

注意，这里动态订阅的 topic，需要在 Kafka 集群中已经存在。如果要动态创建 topic，可以使用 AdminClient API 的 `createTopics()` 方法。例如：

NewTopic newTopic = new NewTopic("test-topic", 3, (short) 1);
CreateTopicsResult result = adminClient.createTopics(Collections.singleton(newTopic));
result.all().get(); // 等待创建完成

kafka consumer 参数

Kafka Consumer 是 Kafka 消费者的客户端 API，可以使用该客户端 API 从 Kafka 集群中读取数据。下面是 Kafka Consumer 常用参数：

1. bootstrap.servers：指定 Kafka 集群的 broker 地址列表，可以是一个或多个地址，多个地址用逗号隔开。

2. group.id：指定消费者所属的消费组 ID，同一个消费组中的消费者共同消费一个 Topic 的消息。

3. enable.auto.commit：指定是否开启自动提交 offset，如果开启，Kafka Consumer 会自动将消费到的最新 offset 提交到 Kafka 集群。

4. auto.commit.interval.ms：设置自动提交 offset 的时间间隔，单位是毫秒。

5. key.deserializer：指定 key 的反序列化器，将字节数组反序列化成 Java 对象。

6. value.deserializer：指定 value 的反序列化器，将字节数组反序列化成 Java 对象。

7. max.poll.records：指定每次 poll 操作最多获取的消息数量。

8. auto.offset.reset：指定当 offset 不存在或者超出范围时的处理方式，可以为 earliest 或 latest。

9. session.timeout.ms：指定消费者与消费组协调器之间的会话超时时间，单位是毫秒。

10. fetch.min.bytes：指定每次 fetch 操作最少获取的字节数。

11. fetch.max.wait.ms：指定每次 fetch 操作最长等待时间，超过该时间仍然没有可用数据，Kafka Consumer 将返回空数据。

12. client.id：指定消费者的 client ID。