Go+Kafka實現延遲消息的實現示例
前言
延遲隊列是一個非常有用的工具,我們經常遇到需要使用延遲隊列的場景,比如延遲通知,訂單關閉等等。
這篇文章主要是使用Go+Kafka實現延遲消息。
使用瞭sarama客戶端。
原理
Kafka實現延遲消息分為下面三步:
- 生產者把消息發送到
延遲隊列
- 延遲服務把
延遲隊列
裡超過延遲時間的消息寫入真實隊列
- 消費者消費
真實隊列
裡的消息
簡單的實現
生產者
生產者隻是把消息發送到延遲隊列
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) }
延遲服務
延遲服務會訂閱延遲隊列
的消息,並把超時消息
發送到真實隊列
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() { // 如果消息已經超時,把消息發送到真實隊列 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 }
消費者
消費者隻是訂閱真實隊列
並消費消息
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 }
改進點
通用的延遲服務
可以把延遲服務封裝成一個通用的服務,這樣生產者可以直接把消息發送給延遲服務,讓延遲服務去處理剩下的邏輯。
延遲服務可以提供多個延時等級,比如5s、10s、30s、1m、5m、10m、1h、2h等,類似於RocketMQ。
生產者負責延遲服務
也可以讓生產者負責延遲服務,讓生產者自己把延遲隊列裡面的消息發送到真實隊列。
下面是一個簡單的實現:
// KafkaDelayQueueProducer 延遲隊列生產者,包含瞭生產者和延遲服務 type KafkaDelayQueueProducer struct { producer sarama.SyncProducer // 生產者 delayTopic string // 延遲服務主題 } // NewKafkaDelayQueueProducer 創建延遲隊列生產者 // producer 生產者 // delayServiceConsumerGroup 延遲服務消費者 // delayTime 延遲時間 // delayTopic 延遲服務主題 // realTopic 真實隊列主題 func NewKafkaDelayQueueProducer(producer sarama.SyncProducer, delayServiceConsumerGroup sarama.ConsumerGroup, delayTime time.Duration, delayTopic, realTopic string) *KafkaDelayQueueProducer { // 啟動延遲服務 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 發送消息 func (q *KafkaDelayQueueProducer) SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) { msg.Topic = q.delayTopic return q.producer.SendMessage(msg) } // DelayServiceConsumer 延遲服務消費者 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() { // 如果消息已經超時,把消息發送到真實隊列 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 }
總結
使用中間隊列
+輪詢
可以很容易的在Kafka實現延遲消息,如果需要一個通用的延遲隊列也可以實現一個通用的延遲服務,也可以讓消費者負責延遲服務的功能。
完整代碼:
- 簡單實現例子:https://github.com/jiaxwu/dq/blob/main/kafka_delay_queue_producer.go
- 包含延遲服務的生產者:https://github.com/jiaxwu/dq/tree/main/kafka_delay_queue_example
到此這篇關於Go+Kafka實現延遲消息的實現示例的文章就介紹到這瞭,更多相關Go Kafka延遲消息內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!