洞悉Kafka：如何巧妙设置消息过期时间与查看存活时间？

后端

2023-06-14 16:07:55

揭秘 Kafka 消息过期时间：增强事件驱动的体系架构

导言

在现代事件驱动的体系架构中，消息传递扮演着至关重要的角色。Kafka 作为领先的消息传递平台，凭借其强大的功能和灵活性，在众多场景中得到了广泛应用。其中一项关键功能便是消息过期时间，它赋予了开发者控制消息有效期和管理缓冲区的能力。

Kafka 消息过期时间

定义：

Kafka 消息过期时间是指消息在主题中保留的最长时间。一旦过期，消息将被删除，无法再被消费。

设置方式：

过期时间可以通过两种方式设置：

在主题创建时设置默认过期时间： 使用 kafka-topics 命令指定 message.timeout.ms 参数。

bin/kafka-topics --create --topic test --config message.timeout.ms=30000

在消息发送时为每个消息单独设置过期时间： 在发送消息时，通过 Timestamp 对象指定过期时间。

producer.send(new ProducerRecord("test", null, null, message.getBytes(), new Timestamp(messageExpirationTime)));

查看消息存活时间

有多种方法可以查看主题下的消息存活时间：

使用 Kafka 命令行工具：

bin/kafka-run-class kafka.tools.GetMessageOffsetsShell --topic test --time -1

使用 Kafka AdminClient：

List<TopicPartition> partitions = Arrays.asList(new TopicPartition("test", 0));
Map<TopicPartition, Long> topicPartitionLongMap = adminClient.listTopicPartitionTimestamps(partitions).get();
Long logEndOffset = topicPartitionLongMap.get(new TopicPartition("test", 0));
long logSize = adminClient.describeLogDirs(Collections.singleton(logDir)).get().values().iterator().next().size();
long expirationHorizonMs = logSize - logEndOffset;

巧妙利用 Kafka 消息过期时间

Kafka 消息过期时间可以发挥多种作用：

延迟队列： 通过设置消息过期时间，可以实现延迟队列。消息在过期前不会被消费，从而达到延迟投递的效果。
缓冲区： Kafka 可用作缓冲区，暂时存储数据。设置消息过期时间可以确保数据不会在缓冲区中停留过久。
事件驱动的体系架构： 在事件驱动的体系架构中，过期时间可用于控制事件的有效期。过期后，事件将不再被处理。

示例

代码示例：

// 设置主题默认过期时间
bin/kafka-topics --create --topic test --config message.timeout.ms=30000

// 创建延迟队列
// 设置消息过期时间为 10 秒
long messageExpirationTime = System.currentTimeMillis() + 10000;
producer.send(new ProducerRecord("test", null, null, message.getBytes(), new Timestamp(messageExpirationTime)));