日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

學(xué)無先后，達(dá)者為師

網(wǎng)站首頁 編程語言 正文

Go+Kafka實(shí)現(xiàn)延遲消息的實(shí)現(xiàn)示例_Golang

作者：jiaxwu ? 更新時(shí)間： 2022-09-16 編程語言

前言

延遲隊(duì)列是一個(gè)非常有用的工具，我們經(jīng)常遇到需要使用延遲隊(duì)列的場(chǎng)景，比如延遲通知，訂單關(guān)閉等等。

這篇文章主要是使用Go+Kafka實(shí)現(xiàn)延遲消息。

使用了sarama客戶端。

原理

Kafka實(shí)現(xiàn)延遲消息分為下面三步：

  • 生產(chǎn)者把消息發(fā)送到延遲隊(duì)列
  • 延遲服務(wù)把延遲隊(duì)列里超過延遲時(shí)間的消息寫入真實(shí)隊(duì)列
  • 消費(fèi)者消費(fèi)真實(shí)隊(duì)列里的消息

簡(jiǎn)單的實(shí)現(xiàn)

生產(chǎn)者

生產(chǎn)者只是把消息發(fā)送到延遲隊(duì)列

msg := &sarama.ProducerMessage{
   Topic: kafka_delay_queue_test.DelayTopic,
   Value: sarama.ByteEncoder("test" + strconv.Itoa(i)),
}
if _, _, err := producer.SendMessage(msg); err != nil {
   log.Println(err)
}

延遲服務(wù)

延遲服務(wù)會(huì)訂閱延遲隊(duì)列的消息，并把超時(shí)消息發(fā)送到真實(shí)隊(duì)列

if err = consumerGroup.Consume(context.Background(),
   []string{kafka_delay_queue_test.DelayTopic}, consumer); err != nil {
   break
}
type Consumer struct {
   producer sarama.SyncProducer
   delay    time.Duration
}

func NewConsumer(producer sarama.SyncProducer, delay time.Duration) *Consumer {
   return &Consumer{
      producer: producer,
      delay:    delay,
   }
}

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
   for message := range claim.Messages() {
      // 如果消息已經(jīng)超時(shí)，把消息發(fā)送到真實(shí)隊(duì)列
      now := time.Now()
      if now.Sub(message.Timestamp) >= c.delay {
         _, _, err := c.producer.SendMessage(&sarama.ProducerMessage{
            Topic: kafka_delay_queue_test.RealTopic,
            Key:   sarama.ByteEncoder(message.Key),
            Value: sarama.ByteEncoder(message.Value),
         })
         if err == nil {
            session.MarkMessage(message, "")
         }
         continue
      }
      // 否則休眠一秒
      time.Sleep(time.Second)
      return nil
   }
   return nil
}

消費(fèi)者

消費(fèi)者只是訂閱真實(shí)隊(duì)列并消費(fèi)消息

if err = consumerGroup.Consume(context.Background(), 
   []string{kafka_delay_queue_test.RealTopic}, consumer); err != nil {
   break
}
type Consumer struct{}

func NewConsumer() *Consumer {
   return &Consumer{}
}

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
   for message := range claim.Messages() {
      fmt.Println("收到消息：", message.Value, message.Timestamp)
      session.MarkMessage(message, "")
   }
   return nil
}

改進(jìn)點(diǎn)

通用的延遲服務(wù)

可以把延遲服務(wù)封裝成一個(gè)通用的服務(wù)，這樣生產(chǎn)者可以直接把消息發(fā)送給延遲服務(wù)，讓延遲服務(wù)去處理剩下的邏輯。

延遲服務(wù)可以提供多個(gè)延時(shí)等級(jí)，比如5s、10s、30s、1m、5m、10m、1h、2h等，類似于RocketMQ。

生產(chǎn)者負(fù)責(zé)延遲服務(wù)

也可以讓生產(chǎn)者負(fù)責(zé)延遲服務(wù)，讓生產(chǎn)者自己把延遲隊(duì)列里面的消息發(fā)送到真實(shí)隊(duì)列。

下面是一個(gè)簡(jiǎn)單的實(shí)現(xiàn)：

// KafkaDelayQueueProducer 延遲隊(duì)列生產(chǎn)者，包含了生產(chǎn)者和延遲服務(wù)
type KafkaDelayQueueProducer struct {
   producer   sarama.SyncProducer // 生產(chǎn)者
   delayTopic string              // 延遲服務(wù)主題
}

// NewKafkaDelayQueueProducer 創(chuàng)建延遲隊(duì)列生產(chǎn)者
// producer 生產(chǎn)者
// delayServiceConsumerGroup 延遲服務(wù)消費(fèi)者
// delayTime 延遲時(shí)間
// delayTopic 延遲服務(wù)主題
// realTopic 真實(shí)隊(duì)列主題
func NewKafkaDelayQueueProducer(producer sarama.SyncProducer, delayServiceConsumerGroup sarama.ConsumerGroup,
   delayTime time.Duration, delayTopic, realTopic string) *KafkaDelayQueueProducer {
   // 啟動(dòng)延遲服務(wù)
   consumer := NewDelayServiceConsumer(producer, delayTime, realTopic)
   go func() {
      for {
         if err := delayServiceConsumerGroup.Consume(context.Background(),
            []string{delayTopic}, consumer); err != nil {
            break
         }
      }
   }()
   return &KafkaDelayQueueProducer{
      producer:   producer,
      delayTopic: delayTopic,
   }
}

// SendMessage 發(fā)送消息
func (q *KafkaDelayQueueProducer) SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) {
   msg.Topic = q.delayTopic
   return q.producer.SendMessage(msg)
}

// DelayServiceConsumer 延遲服務(wù)消費(fèi)者
type DelayServiceConsumer struct {
   producer  sarama.SyncProducer
   delay     time.Duration
   realTopic string
}

func NewDelayServiceConsumer(producer sarama.SyncProducer, delay time.Duration,
   realTopic string) *DelayServiceConsumer {
   return &DelayServiceConsumer{
      producer:  producer,
      delay:     delay,
      realTopic: realTopic,
   }
}

func (c *DelayServiceConsumer) ConsumeClaim(session sarama.ConsumerGroupSession,
   claim sarama.ConsumerGroupClaim) error {
   for message := range claim.Messages() {
      // 如果消息已經(jīng)超時(shí)，把消息發(fā)送到真實(shí)隊(duì)列
      now := time.Now()
      if now.Sub(message.Timestamp) >= c.delay {
         _, _, err := c.producer.SendMessage(&sarama.ProducerMessage{
            Topic: c.realTopic,
            Key:   sarama.ByteEncoder(message.Key),
            Value: sarama.ByteEncoder(message.Value),
         })
         if err == nil {
            session.MarkMessage(message, "")
         }
         continue
      }
      // 否則休眠一秒
      time.Sleep(time.Second)
      return nil
   }
   return nil
}

func (c *DelayServiceConsumer) Setup(sarama.ConsumerGroupSession) error {
   return nil
}

func (c *DelayServiceConsumer) Cleanup(sarama.ConsumerGroupSession) error {
   return nil
}

總結(jié)

使用中間隊(duì)列+輪詢可以很容易的在Kafka實(shí)現(xiàn)延遲消息，如果需要一個(gè)通用的延遲隊(duì)列也可以實(shí)現(xiàn)一個(gè)通用的延遲服務(wù)，也可以讓消費(fèi)者負(fù)責(zé)延遲服務(wù)的功能。

完整代碼：

  • 簡(jiǎn)單實(shí)現(xiàn)例子：https://github.com/jiaxwu/dq/blob/main/kafka_delay_queue_producer.go
  • 包含延遲服務(wù)的生產(chǎn)者：https://github.com/jiaxwu/dq/tree/main/kafka_delay_queue_example

原文鏈接：https://juejin.cn/post/7057584094766432270

欄目分類
最近更新