Kafka消费者如何配置偏移量

在Kafka中，消费者偏移量（Offset）是用来记录消费者在每个分区（Partition）中消费的位置。以下是如何配置Kafka消费者的偏移量：

1. 自动提交偏移量

默认情况下，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"); // 自动提交间隔时间，单位毫秒

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

2. 手动提交偏移量

如果你需要更精细的控制，可以选择手动提交偏移量。手动提交可以在消费完成后显式地调用commitSync或commitAsync方法。

同步提交

try {
    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());
        }
        consumer.commitSync(); // 同步提交偏移量
    }
} finally {
    consumer.close();
}

异步提交

try {
    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());
        }
        consumer.commitAsync((offsets, exception) -> {
            if (exception != null) {
                exception.printStackTrace();
            }
        }); // 异步提交偏移量
    }
} finally {
    consumer.close();
}

3. 从特定偏移量开始消费

你可以指定消费者从特定的偏移量开始消费。这可以通过seek方法实现。

// 假设你已经创建了一个分区为0的消费者
TopicPartition partition = new TopicPartition("your-topic", 0);
long offset = 100L; // 你想从偏移量100开始消费
consumer.assign(Arrays.asList(partition));
consumer.seek(partition, offset);

try {
    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());
        }
    }
} finally {
    consumer.close();
}

4. 从最新的偏移量开始消费

如果你希望消费者从最新的消息开始消费，可以使用seekToEnd方法。

TopicPartition partition = new TopicPartition("your-topic", 0);
consumer.assign(Arrays.asList(partition));
consumer.seekToEnd(partition);

try {
    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());
        }
    }
} finally {
    consumer.close();
}

5. 从最早的偏移量开始消费

如果你希望消费者从最早的可用消息开始消费，可以使用seekToBeginning方法。

TopicPartition partition = new TopicPartition("your-topic", 0);
consumer.assign(Arrays.asList(partition));
consumer.seekToBeginning(partition);

try {
    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());
        }
    }
} finally {
    consumer.close();
}

通过这些配置和方法，你可以灵活地控制Kafka消费者的偏移量，以满足不同的业务需求。