kafka消费命令怎么设置偏移量 - 问答

在Kafka中，消费者可以通过设置偏移量来控制从哪个位置开始消费消息。以下是几种常见的设置偏移量的方法：

自动提交偏移量：

在创建消费者时，可以将enable.auto.commit属性设置为true，这样消费者会在每次消费完一条消息后自动提交偏移量。你可以通过以下代码设置：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000"); // 设置自动提交偏移量的间隔时间，单位毫秒

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

手动提交偏移量：

如果你希望更精细地控制偏移量的提交，可以将enable.auto.commit属性设置为false，然后使用commitSync()或commitAsync()方法手动提交偏移量。以下是一个示例：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");

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

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

从特定偏移量开始消费：

如果你希望从某个特定的偏移量开始消费消息，可以在创建消费者时设置offset.reset属性。例如，如果你希望从偏移量5开始消费，可以将offset.reset设置为earliest（从最早的消息开始）或specific（从指定的偏移量开始）。以下是一个示例：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("offset.reset", "earliest"); // 从最早的消息开始消费

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

请注意，如果你将offset.reset设置为specific，你还需要在消费过程中使用seek()方法来设置当前的偏移量。

0 赞

0 踩