kafka实现延迟消费的代码

Kafka 实现延迟消费的方法是使用 Kafka 的消息延迟功能。具体实现步骤如下：

1. 设置消息延迟时间
在生产者发送消息时，设置消息的延迟时间。可以使用 Kafka 提供的 `ProducerRecord` 类的 `headers()` 方法设置消息头部信息，示例代码如下：

long delay = 10000; // 延迟10秒
ProducerRecord<String, String> record = new ProducerRecord<String, String>("test-topic", "key", "value");
record.headers().add(new RecordHeader("delay", BytesUtils.longToBytes(delay)));
producer.send(record);

上述代码中，通过设置消息头部信息中的 `delay` 字段来实现消息延迟。

2. 创建消费者
创建消费者，订阅消息，并在消费消息时判断消息的延迟时间是否已经到达。如果未到达延迟时间，则忽略该消息，示例代码如下：

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("test-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        long delay = 0;
        for (Header header : record.headers()) {
            if (header.key().equals("delay")) {
                delay = BytesUtils.bytesToLong(header.value());
            }
        }
        if (System.currentTimeMillis() - delay > record.timestamp()) {
            System.out.println("消费消息：" + record.value());
        } else {
            System.out.println("延迟消息：" + record.value());
        }
    }
}

上述代码中，使用 `ConsumerRecords` 对象获取消费到的消息，然后遍历消息列表，获取消息头部信息中的 `delay` 字段，判断是否已经到达延迟时间。如果到达延迟时间，则消费消息，否则忽略该消息。

需要注意的是，上述代码中使用了 `BytesUtils` 工具类来对字节数组和 long 类型进行转换，这个工具类需要自己实现。另外，在生产者发送消息时，需要设置 Kafka 的消息延迟时间配置项 `linger.ms` 和 `delivery.timeout.ms`。这两个配置项可以设置为相同的值，表示消息延迟时间。