Go+Kafka实现消息的过期时间（TTL）

前言

有时候我们需要带过期时间的消息，然而Kafka消息并不自带TTL选项，因此我们需要自己实现。

这篇文章主要是使用Go实现Kafka带过期时间的消息。

使用了sarama客户端。

原理

其主要原理就是在消费者端使用拦截器过滤掉已经过期的消息。

固定过期时间

如果消息的过期时间是固定的，可以简单的在生产者端设置消息的Timestamp字段为当前时间，然后在消费者端根据消息的Timestamp字段和过期时间判断消息是否过期。

生产者端

我们设置消息的Timestamp字段为当前时间，这样消费者就可以知道消息是什么时候发送的。

msg := &sarama.ProducerMessage{
   Topic:     fix_ttl_test.Topic,
   Value:     sarama.ByteEncoder(fix_ttl_test.Topic + strconv.Itoa(i)),
   // 设置为当前时间
   Timestamp: time.Now(),
}
if _, _, err := producer.SendMessage(msg); err != nil {
   log.Println(err)
}

消费者端

我们在消息被消费时添加一段根据消息的Timestamp和过期时间判断消息是否过期的逻辑，把过期的消息丢弃掉。

type FixTTLConsumer struct {
   ttl time.Duration // 消息过期时间
}

func NewFixTTLConsumer(ttl time.Duration) *FixTTLConsumer {
   return &FixTTLConsumer{ttl: ttl}
}

func (c *FixTTLConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
   for message := range claim.Messages() {
      // 判断消息是否过期
      now := time.Now()
      if now.Sub(message.Timestamp) >= c.ttl {
         // 过期则丢弃消息
         fmt.Println("消息过期：", message.Value, message.Timestamp)
         session.MarkMessage(message, "")
         continue
      }

      // 有效的消息
      fmt.Println("消息有效：", message.Value, message.Timestamp)
      session.MarkMessage(message, "")
   }
   return nil
}

自定义过期时间

如果消息的过期时间是不固定的，可以在生产者端设置消息的Timestamp字段为当前时间，在Headers字段里设置消息的过期时间，然后在消费者端根据消息的Timestamp字段和消息的过期时间判断消息是否过期。

生产者端

我们设置消息的Timestamp字段为当前时间，在消息Headers字段里设置消息的过期时间，这样消费者就知道消息是什么时候发送的，并且知道过期时间是多少。

msg := &sarama.ProducerMessage{
   Topic: custom_ttl_test.Topic,
   Value: sarama.ByteEncoder(custom_ttl_test.Topic + strconv.Itoa(i)),
   // 设置为当前时间
   Timestamp: time.Now(),
   // 把消息的过期时间添加到消息的Headers字段里
   Headers: []sarama.RecordHeader{{
      Key:   sarama.ByteEncoder("ttl"),
      Value: int64ToBytes(int64(time.Second * time.Duration(i))),
   }},
}
if _, _, err := producer.SendMessage(msg); err != nil {
   log.Println(err)
}

消费者端

我们在消息被消费时添加一段根据消息的Timestamp和过期时间判断消息是否过期的逻辑，把过期的消息丢弃掉。

和上面固定时间的代码基本相同，只是这里的ttl是从消息的Headers里面获取。

type CustomTTLConsumer struct{}

func NewCustomTTLConsumer() *CustomTTLConsumer {
   return &CustomTTLConsumer{}
}

func (c *CustomTTLConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
   for message := range claim.Messages() {
      // 判断消息是否过期
      // 从消息的Headers字段获取ttl
      var ttl time.Duration
      for _, header := range message.Headers {
         if bytes.Equal(header.Key, []byte("ttl")) {
            ttl = time.Duration(binary.BigEndian.Uint64(header.Value))
            break
         }
      }
      now := time.Now()
      if now.Sub(message.Timestamp) >= ttl {
         // 过期则丢弃消息
         fmt.Println("消息过期：", message.Value, message.Timestamp)
         session.MarkMessage(message, "")
         continue
      }

      // 有效的消息
      fmt.Println("消息有效：", message.Value, message.Timestamp)
      session.MarkMessage(message, "")
   }
   return nil
}

总结

Kafka实现消息的过期时间主要是通过在消费者端对消息是否过期进行判断，而生产者端需要设置Timestamp字段或Headers字段为消费者端提供过期时间判断的依据。

完整示例代码：github.com/jiaxwu/kttl

参考