// 0 配置
Properties properties = new Properties();
// 連接 bootstrap.servers
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");
// 反序列化
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 配置消費(fèi)者組id
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test5");
// 1 創(chuàng)建一個(gè)消費(fèi)者  "", "hello"
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
// 2 訂閱主題 first
ArrayList<String> topics = new ArrayList<>();
topics.add("first");
kafkaConsumer.subscribe(topics);
// 3 消費(fèi)數(shù)據(jù)
while (true){
    //1秒消費(fèi)一次
    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        System.out.println(consumerRecord);
    }
}

// 0 配置
Properties properties = new Properties();
// 連接
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");
// 反序列化
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 組id
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
// 1 創(chuàng)建一個(gè)消費(fèi)者
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
// 2 訂閱主題對(duì)應(yīng)的分區(qū)
ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
topicPartitions.add(new TopicPartition("first",0));
kafkaConsumer.assign(topicPartitions);
// 3 消費(fèi)數(shù)據(jù)
while (true){
    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        System.out.println(consumerRecord);
    }
}

properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");

// 自動(dòng)提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
// 提交時(shí)間間隔
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);

// 0 配置
Properties properties = new Properties();
// 連接 bootstrap.servers
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");
// 反序列化
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 配置消費(fèi)者組id
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
// 手動(dòng)提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
// 1 創(chuàng)建一個(gè)消費(fèi)者  "", "hello"
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
// 2 訂閱主題 first
ArrayList<String> topics = new ArrayList<>();
topics.add("first");
kafkaConsumer.subscribe(topics);
// 3 消費(fèi)數(shù)據(jù)
while (true){
    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        System.out.println(consumerRecord);
    }
    // 同步提交 offset
    kafkaConsumer.commitSync();
}

// 0 配置
Properties properties = new Properties();
// 連接 bootstrap.servers
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");
// 反序列化
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 配置消費(fèi)者組id
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
// 手動(dòng)提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
// 1 創(chuàng)建一個(gè)消費(fèi)者  "", "hello"
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
// 2 訂閱主題 first
ArrayList<String> topics = new ArrayList<>();
topics.add("first");
kafkaConsumer.subscribe(topics);
// 3 消費(fèi)數(shù)據(jù)
while (true){
    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        System.out.println(consumerRecord);
    }
    // 同步提交 offset
    kafkaConsumer.commitAsync();
}

// 0 配置信息
Properties properties = new Properties();
// 連接
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");
// 反序列化
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 組id
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test3");
// 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ū)分配方案已經(jīng)制定完畢
while (assignment.size() == 0){
    kafkaConsumer.poll(Duration.ofSeconds(1));
    assignment = kafkaConsumer.assignment();
}
// 指定消費(fèi)的offset
for (TopicPartition topicPartition : assignment) {
    kafkaConsumer.seek(topicPartition,600);
}
// 3  消費(fèi)數(shù)據(jù)
while (true){
    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        System.out.println(consumerRecord);
    }
}

// 0 配置信息
Properties properties = new Properties();
// 連接
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");
// 反序列化
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 組id
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test3");
// 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ū)分配方案已經(jīng)制定完畢
while (assignment.size() == 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) {
   topicPartitionLongHashMap.put(topicPartition,System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
}
Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = kafkaConsumer.offsetsForTimes(topicPartitionLongHashMap);
// 指定消費(fèi)的offset
for (TopicPartition topicPartition : assignment) {
   OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition);
   kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
}
// 3  消費(fèi)數(shù)據(jù)
while (true){
   ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
   for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
       System.out.println(consumerRecord);
   }
}