Kafka技术问询：基于Offset消费数据、生产者获Offset及精准查询

Hey there! Let's break down your Kafka questions one by one with practical explanations and code examples:

1. 如何使用Offset让Kafka Consumer获取数据？ #

Kafka的Offset是每个分区（Partition）中消息的唯一标识，消费者通过Offset来跟踪自己已经消费到的位置。要利用Offset控制消费，主要有几种常见场景：

• 从头开始消费所有消息：如果想从分区的起始位置（Offset 0）开始消费，可以在订阅主题后调用seekToBeginning()方法：

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));

// 重置到每个分区的起始Offset
consumer.seekToBeginning(consumer.assignment());

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());
    }
}

• 从指定Offset开始消费：如果需要精确跳到某个分区的特定Offset，先获取分区列表，再调用seek()方法：

// 订阅主题后，先获取当前分配的分区集合
consumer.subscribe(Collections.singletonList("my-topic"));
Set<TopicPartition> partitions = consumer.assignment();
// 等待消费者完成分区分配（刚订阅时可能需要短暂等待）
consumer.poll(Duration.ofMillis(100));
partitions = consumer.assignment();

// 针对每个分区设置目标Offset
for (TopicPartition partition : partitions) {
    consumer.seek(partition, 100); // 跳到该分区的Offset 100位置
}

// 开始消费
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    // 处理记录...
}

• 从最新位置消费：如果只想消费订阅之后新产生的消息，可以调用seekToEnd()方法，原理和上面类似。

2. 向Kafka Producer写入数据时，能否获取记录的Offset？ #

当然可以！当Producer发送消息后，Kafka Broker会返回RecordMetadata对象，里面包含了这条消息所在的Partition、Offset、Timestamp等关键信息。你可以通过两种方式获取：

• 同步发送获取：使用send()方法的返回值Future<RecordMetadata>，调用get()阻塞等待结果：

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "value1");

try {
    RecordMetadata metadata = producer.send(record).get();
    System.out.printf("消息已发送到分区 %d，Offset为 %d%n", metadata.partition(), metadata.offset());
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
} finally {
    producer.close();
}

• 异步回调获取：如果不想阻塞，可以在send()方法中传入Callback接口，在回调函数中处理返回的Metadata：

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.printf("异步回调：消息分区 %d，Offset %d%n", metadata.partition(), metadata.offset());
    } else {
        exception.printStackTrace();
    }
});

3. 是否可以使用相同的Offset和Partition检索特定记录？ #

是的，但有个前提：这条记录还没有被Kafka的日志清理策略删除（比如超过了保留时间、日志大小达到阈值等）。只要记录还在Broker上，你可以通过指定Partition和Offset来精准获取它。

示例代码如下：

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// 直接指定要消费的分区，而不是订阅整个主题
TopicPartition targetPartition = new TopicPartition("my-topic", 0);
consumer.assign(Collections.singletonList(targetPartition));

// 跳到目标Offset
long targetOffset = 100;
consumer.seek(targetPartition, targetOffset);

// 只拉取一条记录（因为我们只需要指定Offset的那条）
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    if (record.offset() == targetOffset) {
        System.out.printf("找到目标记录：Offset %d，Value %s%n", record.offset(), record.value());
        break;
    }
}

consumer.close();

需要注意的是，如果指定的Offset已经不存在（比如被清理了），消费者会根据auto.offset.reset配置的策略来处理：默认是latest（跳到最新Offset），也可以设置为earliest（跳到起始Offset）或者none（直接抛出异常）。

内容的提问来源于stack exchange，提问作者Beginner