[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic atguigu --partitions 2 --replication-factor 2

[atguigu@hadoop102 kafka] $ bin/kafka-console-producer.sh --topic atguigu --bootstrap-server hadoop102:9092

[atguigu@hadoop104 kafka]$ bin/kafka-console-consumer.sh bootstrap-server hadoop102:9092--topic atguigu --group test

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --topic _consumer_offsets --bootstrap-server hadoop102:9092 --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning

//配置是否是自動(dòng)提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
//提交時(shí)間間隔，單位是ms
properties.put(ConsumerConfig.AUTO_COMNIT_INTERVAL_NS_CONFI6,1000);

//手動(dòng)提交屬性配置
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG ,false);
//消費(fèi)代碼邏輯
XXX
XXX
XXX
//手動(dòng)提交代碼(處理完數(shù)據(jù)以后，這里為了方便，只展示關(guān)鍵代碼)
//手動(dòng)提交offset
kafkaConsumer.commitsync();

//手動(dòng)提交屬性配置
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG ,false);
//消費(fèi)代碼邏輯
XXX
XXX
XXX
//手動(dòng)提交代碼(處理完數(shù)據(jù)以后，這里為了方便，只展示關(guān)鍵代碼)
//手動(dòng)提交offset
kafkaConsumer.commitAsync();

//1創(chuàng)建消費(fèi)者
KafkaConsumer<String，String> kafkaConsumer = new KafkaConsumer<>(properties);
// 2訂閱主題
ArrayList<String> topics = new ArrayList<>(;topics.add( "first");
kafkaConsumer.subscribe(topics);
 
//指定位置進(jìn)行消費(fèi)
set<TopicPartition> assignment = kafkaConsumer.assignment();//獲取所有分區(qū)信息
//保證分區(qū)分配方案已經(jīng)制定完畢，因?yàn)橛捎趌eader消費(fèi)者制定分配方案會(huì)消耗一定時(shí)間，有可能此時(shí)獲取不到分區(qū)信息,所以加一層分區(qū)空間判斷
while (assignment.size() == 0){
    //促使獲取的分區(qū)數(shù)量不為0
    kafkaConsumer.poll(Duration.ofSeconds(1));
    assignment = kafkaConsumer.assignment();
}
 
//遍歷所有分區(qū)，指定消費(fèi)的offset
for (TopicPartition topicPartition : assignment) {
    kafkaConsumer.seek(topicPartition, 100);
}
 
// 3消費(fèi)數(shù)據(jù)
while (true){

//1創(chuàng)建消費(fèi)者
KafkaConsumer<String，String> kafkaConsumer = new KafkaConsumer<>(properties);
// 2訂閱主題
ArrayList<String> topics = new ArrayList<>(;topics.add( "first");
kafkaConsumer.subscribe(topics);
 
//指定位置進(jìn)行消費(fèi)
set<TopicPartition> assignment = kafkaConsumer.assignment();//獲取所有分區(qū)信息
//保證分區(qū)分配方案已經(jīng)制定完畢，因?yàn)橛捎趌eader消費(fèi)者制定分配方案會(huì)消耗一定時(shí)間，有可能此時(shí)獲取不到分區(qū)信息,所以加一層分區(qū)空間判斷
while (assignment.size() == 0){
    //促使獲取的分區(qū)數(shù)量不為0
    kafkaConsumer.poll(Duration.ofSeconds(1));
    assignment = kafkaConsumer.assignment();
}
//希望把時(shí)間轉(zhuǎn)換為對(duì)應(yīng)的offset
HashMap<TopicPartition，Long> topicPartitionLongHashMap = new HashMap<>();
//封裝對(duì)應(yīng)集合
for (TopicPartition topicPartition : assignment) {
    //希望獲取當(dāng)前系統(tǒng)時(shí)間一天前的數(shù)據(jù)。
    topicPartitionLongHashMap.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
}
Nap<TopicPartition，OffsetAnd imestamp> topioPartitionffsetAndrtimestampMep = karfiaConsumer.offsetsForTines(topicPartitionL ongHashiap);
 
 
//遍歷所有分區(qū)，指定消費(fèi)的offset
//指定消費(fèi)的offset
for (TopicPartition topicPartition : assignment) {
    OffsetAndTimestamp offsetAndTimestamp = topicPartition0ffsetAndTimestampHap.get(topicPartition);
    kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
}
 
// 3消費(fèi)數(shù)據(jù)
while (true){