kafka指定消费1分钟前的数据实现延迟消费

可以通过Kafka的时间戳特性来实现延迟消费。具体做法是，将消息的时间戳设置为发送时间，然后在消费者端使用Kafka的时间戳消费功能，只消费指定时间戳之前的消息。具体步骤如下：

1. 在生产者端，将消息的时间戳设置为当前时间戳，即System.currentTimeMillis()。

2. 在消费者端，使用Kafka提供的时间戳消费功能，通过设置ConsumerConfig#AUTO_OFFSET_RESET_CONFIG参数为"timestamp"，并将ConsumerConfig#TIMESTAMP_CONFIG参数设置为指定的时间戳，即可只消费指定时间戳之前的消息。

示例代码：

// 生产者端
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
record.timestamp(System.currentTimeMillis());
producer.send(record);

// 消费者端
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "timestamp");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.TIMESTAMP_CONFIG, System.currentTimeMillis() - 60000); // 消费1分钟前的数据
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    System.out.printf("offset = %d, key = %s, value = %s, timestamp = %d%n", record.offset(), record.key(), record.value(), record.timestamp());
}

注意：时间戳消费功能需要Kafka版本在0.10.1.0及以上。