ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));中的Duration为什么会显红

`Duration.ofMillis(100)` 是一个Java 8引入的时间间隔表示方法，它用于创建一个持续特定毫秒数的`Duration`对象。在这个上下文中，`Duration.ofMillis(100)` 表示等待Kafka消费者的poll操作最多100毫秒，以便从Kafka主题中获取新的消息记录。

如果你看到这个代码片段在某些IDE（如IntelliJ IDEA）中显示为红色，那可能是由于几个原因：

1. **缺少导入**: 如果IDE提示找不到`Duration`，那可能是因为你还没有导入`java.time.Duration`包，需要添加相应的import语句。

import java.time.Duration;

2. **类型不匹配**: 如果IDE认为这里的类型不适用于当前上下文，比如你可能在一个旧版本的API或错误的项目配置下，那么可能需要更新到支持Java 8时间日期API的版本。
3. **编译错误**: 如果是在运行阶段出错，那可能是其他代码导致的问题，比如该`kafkaConsumer`变量未初始化或`poll()`方法调用有误。

为了确认这个问题，你可以检查一下周围的代码环境，看是否有正确的导入、API兼容性和变量是否已正确设置。如果不确定，可以尝试运行或重新构建项目看看是否会报具体的编译或运行时错误信息。

相关问题

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));这句代码每部分的含义

这段代码使用 Kafka Consumer API 从 Kafka 集群中消费消息，每次调用 `poll` 方法会从 Kafka 中拉取一批消息，返回一个 `ConsumerRecords` 对象，其中包含了一组 `ConsumerRecord` 对象。每个 `ConsumerRecord` 对象表示一条消息，包含了消息的内容和元数据。

具体来说，这段代码的含义如下：

- `consumer`：Kafka Consumer 对象，用于从 Kafka 集群中消费消息。
- `poll(Duration.ofMillis(1000))`：从 Kafka 集群中拉取消息，等待时间为 1000 毫秒。如果在等待时间内没有拉取到任何消息，则返回空的 `ConsumerRecords` 对象。
- `ConsumerRecords<String, String>`：表示一组消息，其中的泛型参数 `<String, String>` 分别表示消息的键和值的类型，这里都是字符串类型。
- `records`：表示从 Kafka 集群中拉取到的消息，是一个 `ConsumerRecords` 对象。

kafka消费不到数据 public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "192.168.56.101:9092"); // Kafka地址 // 每个消费者分配独立的组号 props.put("group.id", "consumer01"); // 如果value合法，则自动提交偏移量 props.put("enable.auto.commit", "true"); // 设置多久一次更新被消费消息的偏移量 props.put("auto.commit.interval.ms", "1000"); // 设置会话响应的时间，超过这个时间kafka可以选择放弃消费或者消费下一条消息 props.put("session.timeout.ms", "30000"); // 指定CURRENT-OFFSET的数值， 自动重置offset, 当读取的 offset 不在当前 kafka topic 范围内，将会从头开始读取 props.put("auto.offset.reset", "earliest"); // latest earliest 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); String topic = "mrt"; //mrt-java consumer.subscribe(Collections.singletonList(topic)); ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

这段代码中只有一次调用了 `consumer.poll()` 方法获取消息，而没有在循环中不断地调用该方法，因此只能消费到一次消息。你需要在一个循环中不断地调用 `consumer.poll()` 方法，以便持续地获取新消息。你可以参考以下代码示例：

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

该代码会持续地从 Kafka 中获取消息，并输出每条消息的偏移量、键和值。你可以根据实际需求，对消息做出相应的处理。