kafka如何实现延迟消费

Kafka 是一个高性能、分布式、可伸缩、可靠的消息系统，它将消息发布和订阅作为一种通信模式，通过以消息为中心的方式实现应用程序之间的通信。在 Kafka 中，消息是以 topic 的形式组织的，而每个 topic 包含了若干个 partition，每个 partition 又包含了若干个 message。

在实际应用中，我们可能会遇到需要延迟消费消息的情况，比如某些业务场景下需要等待一定时间之后才能处理消息。Kafka 提供了两种方式来实现延迟消费：等待时间和时间戳。

1. 等待时间

在使用 Kafka 消费消息时，可以通过指定等待时间的方式来实现延迟消费。具体实现方式是在 Consumer 的 poll() 方法中设置超时时间，如果超时时间内没有新消息到达，则 poll() 方法会返回一个空的消息列表。示例代码如下：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    if (records.count() == 0) {
        // 如果没有新数据，则等待 1 秒钟后继续轮询
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        continue;
    }
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
    }
}

在上述代码中，我们通过调用 Consumer 的 poll() 方法，并设置等待时间为 1 秒钟。如果在 1 秒钟内没有新消息到达，则继续等待 1 秒钟并继续轮询。这样可以实现延迟消费的效果。

2. 时间戳

除了等待时间外，Kafka 还支持通过指定时间戳的方式来实现延迟消费。