经常，业务上我们遇到消费失败问题，但是呢又不好把失败的所有消息挨个挨个找出来消费，这时候我们就可以采用新建消费端，从指定的时间戳去消费一段时间内的消息来补数据的，这里需要我们消费端满足幂等条件，可以重复消费。

    1. public class ConsumerDemo {
    3.     public static void main(String[] args) throws ParseException {
    5.         String topic = "test_topic01";
    6.         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    7.         Date date = sdf.parse("2022-04-09 12:10:45.617");
    8.         Long end = sdf.parse("2022-04-09 12:10:45.617").getTime();
    9.         // 
    10.         reconsume(topic, date, end);
    11.     }
    13.     // 从指定的时间开始消费kafka消息
    14.     private static void reconsume(String topic, Date start, Long end) {
    15.         // 消费端配置
    16.         Properties prop = clientConf();
    17.         // 获取元数据
    18.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop);
    19.         // 拉取分片信息
    20.         List<PartitionInfo> partitions = consumer.partitionsFor(topic);
    21.         // 业务线程池，每个分片单独一个线程处理
    22.         ExecutorService executorService = Executors.newFixedThreadPool(partitions.size());
    23.         // 拉取指定时间戳对应的offset
    24.         Map<TopicPartition, Long> map = new HashMap<>();
    25.         for (PartitionInfo partition : partitions) {
    26.             map.put(new TopicPartition(topic, partition.partition()), start.getTime());
    27.         }
    28.         Map<TopicPartition, OffsetAndTimestamp> timestampMap = consumer.offsetsForTimes(map);
    29.         // future集合
    30.         List<CompletableFuture> list = new ArrayList<>(partitions.size());
    31.         try {
    32.             // 遍历所有分片
    33.             for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : timestampMap.entrySet()) {
    34.                 // 根据
    35.                 CompletableFuture<ConsumerRecord<String, String>> future =
    36.                         run(prop, end, executorService, entry);
    37.                 list.add(future);
    38.             }
    39.             // 等待所有分片消费完成
    40.             CompletableFuture.allOf(list.toArray(new CompletableFuture[0])).join();
    42.             System.out.println(topic + "消费完成");
    43.         } finally {
    44.             executorService.shutdown();
    45.             consumer.close();
    46.         }
    47.     }
    49.     private static CompletableFuture<ConsumerRecord<String, String>> run(Properties prop, Long end, ExecutorService executorService,
    50.                                                                          Map.Entry<TopicPartition, OffsetAndTimestamp> entry) {
    51.         CompletableFuture<ConsumerRecord<String, String>> future = CompletableFuture.supplyAsync(() -> {
    52.             // 消费客户端
    53.             KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop);
    54.             // 分配分片
    55.             consumer.assign(Collections.singleton(entry.getKey()));
    56.             // 根据时间戳转换为offset，指定offset位置开始消费
    57.             consumer.seek(entry.getKey(), entry.getValue().offset());
    58.             while (true) {
    59.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));
    60.                 for (ConsumerRecord<String, String> record : records) {
    62.                     // 消费到指定时间戳，结束
    63.                     if (record.timestamp() > end) {
    64.                         System.out.printf("分区[%d]消费完成[offset=%d]...............\n", record.partition(), record.offset());
    65.                         return record;
    66.                     }
    67.                     System.out.printf("offset=%d,key=%s,val=%s,partition=%d\n", record.offset(), record.key(), record.value(), record.partition());
    68.                 }
    69.             }
    70.         }, executorService);
    71.         return future;
    72.     }
    74.     private static Properties clientConf() {
    75.         Properties prop = new Properties();
    76.         prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.127.157:9092");
    77.         prop.put(ConsumerConfig.GROUP_ID_CONFIG, "group0x");
    78.         prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
    79.         prop.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
    80.         prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    81.         prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    82.         prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    83.         return prop;
    84.     }
    85. }

    为什么可以按时间戳获取消息呢，实际上kafka服务端维护了一个时间戳与offset的对应关系的索引。
    image.png
    在broker数据存储端，维护了一个索引文件，可以根据时间戳找到对应的offset，然后我们可以根据具体的offset去消费消息，这里需要保证broker服务端的服务器时间的正确性。