1 場景描述
在Java中，使用Kafka時祟滴，如果你想從頭開始消費（始終從最早的消息開始消費）振惰，你需要設(shè)置Kafka消費者的auto.offset.reset屬性為earliest。這樣踱启，當(dāng)消費者沒有為分區(qū)找到存在的偏移量時报账，它會從分區(qū)的開始處開始消費消息。
但我使用的是舊的分區(qū)埠偿，已經(jīng)存在分區(qū)的offset透罢，則不會從頭消費。
本博客將從重置分區(qū)offset下手冠蒋， 在不更換消費組的情況設(shè)置topic永久從頭消費羽圃。
2 實現(xiàn)思路
常規(guī)辦法，隨機消費組抖剿，每次測試時候生成一個新的消費組朽寞。
重置offset，遍歷該主題改消費組的分區(qū)信息斩郎，一一重置為0脑融，再去消費數(shù)據(jù)，就實現(xiàn)了topic永久從頭消費缩宜。
3 實現(xiàn)核心代碼
// 獲取分區(qū)信息
List<PartitionInfo> partitions = consumer.partitionsFor("test_topic");
// 因為 seek（） 是懶依賴 需要虛擬調(diào)用一下poll（）方法
ConsumerRecords<String, String> qq = consumer.poll(100);
for (PartitionInfo partition : partitions) {
TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition());
// 指定重置分區(qū)的offset為0
consumer.seek(topicPartition, 0);
}
//提交offset信息
consumer.commitSync()
溫馨提示：因為 seek（） 是懶依賴 肘迎，所以需要虛擬調(diào)用一下poll（）方法。不然會報錯,如下锻煌。
Exception in thread "main" java.lang.IllegalStateException: No current assignment for partition test_topic-0
at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedState(SubscriptionState.java:356)
at org.apache.kafka.clients.consumer.internals.SubscriptionState.seekUnvalidated(SubscriptionState.java:373)
at org.apache.kafka.clients.consumer.KafkaConsumer.seek(KafkaConsumer.java:1569)
at qiyun_test_kafkaConsumer.main(qiyun_test_kafkaConsumer.java:40)
4 kafka永久重頭消費demo
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
public class qiyun_test_kafkaConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "kafak01:9092");
props.put("group.id", "test1");
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");
props.put("auto.offset.reset", "earliest"); // 設(shè)置從最早的位置開始消費
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test_topic"));
// 獲取分區(qū)信息
List<PartitionInfo> partitions = consumer.partitionsFor("test_topic");
// 因為 seek（） 是懶依賴 需要虛擬調(diào)用一下poll（）方法
ConsumerRecords<String, String> qq = consumer.poll(100);
for (PartitionInfo partition : partitions) {
TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition());
// 指定重置分區(qū)的offset為0
consumer.seek(topicPartition, 0);
}
//提交offset信息
consumer.commitSync();
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
}
}
}
}
4 kafka shell 消費命令
./kafka-console-consumer.sh --bootstrap-server kafka01:19092 --topic test_topic --from-beginning
