java自己手动控制kafka的offset操作

当使用kafka作为消费者时，消费者往往需要对消费的offset进行管理，以确保以后能够正确地读取数据。我们通常使用kafka内置的自动提交offset机制，但有时候我们也需要手动控制offset。

下面是一些步骤和示例，让你更好地了解如何手动控制kafka的offset操作：

步骤1：创建kafka消费者

首先，我们需要创建kafka消费者。以下是创建一个简单的kafka消费者的代码段：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
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(Collections.singletonList("test-topic"));

步骤2：手动控制offset

我们通过以下步骤来手动控制offset：

1. 通过调用consumer.poll()方法从kafka取回数据。
2. 根据业务逻辑处理数据。
3. 调用commitSync()方法在处理完所有消息后提交offset。

以下代码显示了如何在消费数据之后手动提交offset：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理业务逻辑
    }
    // 手动提交offset
    consumer.commitSync();
}

可以通过调用commitSync()方法将最新的offset提交到broker。如果需要提交特定的offset，可以调用commitSync(offsets)方法并将自定义的offset作为参数传递：

import org.apache.kafka.common.TopicPartition;

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理业务逻辑
    }
    // 提交特定的offset
    TopicPartition topicPartition = new TopicPartition("test-topic", 0);
    OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(100);
    consumer.commitSync(Collections.singletonMap(topicPartition, offsetAndMetadata));
}

示例1：从指定的offset开始消费

有时候，我们需要针对特定的offset重新消费，下面是一个使用seek方法实现从指定的offset开始消费的示例：

TopicPartition topicPartition = new TopicPartition("test-topic", 0);
consumer.assign(Collections.singletonList(topicPartition));
consumer.seek(topicPartition, 100); // 从100的offset开始消费

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理业务逻辑
    }
    // 手动提交offset
    consumer.commitSync();
}

示例2：手动控制offset并处理异常情况

当消费数据时，有可能会发生异常。在这种情况下，我们需要手动控制offset并决定是重新消费还是忽略损坏的数据。

以下代码片段演示了如何处理异常和手动提交offset：

while (true) {
    try {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // 处理业务逻辑
        }
        // 手动提交offset
        consumer.commitSync();
    } catch (Exception ex) {
        // 处理异常
        consumer.seekToBeginning(Collections.singletonList(new TopicPartition("test-topic", 0)));
    }
}

在上面的示例中，如果在处理消息时发生异常，则会调用seekToBeginning() 方法，并重新消费所有消息。

希望上述步骤和示例能够帮助你成功控制kafka的offset操作。