Kafka中的Consumer level offset和Consumer group offset之间的区别

在Kafka中，Consumer Level Offset和Consumer Group Offset是两种不同的偏移量管理机制，用于消费者在消费消息时跟踪其进度。

Consumer Level Offset是指每个消费者实例在消费消息时，通过维护自己的offset来跟踪消费的进度。每个消费者实例都有自己的offset，并独立地从指定的topic partition读取消息。这意味着每个消费者实例都可以独立地消费消息，无需考虑其他消费者实例。

Consumer Group Offset是指一组消费者实例共享的offset。在一个消费者组中，每个消费者实例会被分配到一个或多个topic partition，每个分区只能由一个消费者实例消费。消费者组中的所有实例共享相同的offset，这意味着它们协同工作，确保每个分区只有一个消费者实例在消费。

下面是一个示例代码，展示了如何使用Consumer Level Offset和Consumer Group Offset：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 配置消费者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-consumer-group");
        props.put("enable.auto.commit", "false"); // 关闭自动提交offset

        // 创建消费者实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅topic
        consumer.subscribe(Arrays.asList("my-topic"));

        // 消费消息
        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 Level Offset
                    consumer.commitSync(Collections.singletonMap(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)));
                }
            }
        } finally {
            consumer.close();
        }
    }
}

在上面的代码中，我们创建了一个消费者实例，并订阅了一个名为"my-topic"的topic。在消费消息时，我们使用了Consumer Level Offset来跟踪消费的进度，并在处理完每条消息后手动提交offset。

如果要使用Consumer Group Offset，只需要将props.put("group.id", "my-consumer-group");中的"group.id"设置为相同的值，并删除props.put("enable.auto.commit", "false");行。这样，消费者实例将自动使用Consumer Group Offset来管理offset，并由Kafka自动提交offset。