第Go+Kafka實現(xiàn)延遲消息的實現(xiàn)示例目錄前言原理簡單的實現(xiàn)生產(chǎn)者延遲服務消費者改進點通用的延遲服務生產(chǎn)者負責延遲服務總結

前言

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

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

使用了sarama客戶端。

原理

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

生產(chǎn)者把消息發(fā)送到延遲隊列延遲服務把延遲隊列里超過延遲時間的消息寫入真實隊列消費者消費真實隊列里的消息

簡單的實現(xiàn)

生產(chǎn)者

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

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)

}

延遲服務

延遲服務會訂閱延遲隊列的消息，并把超時消息發(fā)送到真實隊列

iferr=consumerGroup.Consume(context.Background(),

[]string{kafka_delay_queue_test.DelayTopic},consumer);err!=nil{

break

}

typeConsumerstruct{

producersarama.SyncProducer

delaytime.Duration

funcNewConsumer(producersarama.SyncProducer,delaytime.Duration)*Consumer{

returnConsumer{

producer:producer,

delay:delay,

func(c*Consumer)ConsumeClaim(sessionsarama.ConsumerGroupSession,claimsarama.ConsumerGroupClaim)error{

formessage:=rangeclaim.Messages(){

//如果消息已經(jīng)超時，把消息發(fā)送到真實隊列

now:=time.Now()

ifnow.Sub(message.Timestamp)=c.delay{

_,_,err:=ducer.SendMessage(sarama.ProducerMessage{

Topic:kafka_delay_queue_test.RealTopic,

Key:sarama.ByteEncoder(message.Key),

Value:sarama.ByteEncoder(message.Value),

iferr==nil{

session.MarkMessage(message,"")

continue

//否則休眠一秒

time.Sleep(time.Second)

returnnil

returnnil

}

消費者

消費者只是訂閱真實隊列并消費消息

iferr=consumerGroup.Consume(context.Background(),

[]string{kafka_delay_queue_test.RealTopic},consumer);err!=nil{

break

}

typeConsumerstruct{}

funcNewConsumer()*Consumer{

returnConsumer{}

func(c*Consumer)ConsumeClaim(sessionsarama.ConsumerGroupSession,claimsarama.ConsumerGroupClaim)error{

formessage:=rangeclaim.Messages(){

fmt.Println("收到消息：",message.Value,message.Timestamp)

session.MarkMessage(message,"")

returnnil

}

改進點

通用的延遲服務

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

延遲服務可以提供多個延時等級，比如5s、10s、30s、1m、5m、10m、1h、2h等，類似于RocketMQ。

生產(chǎn)者負責延遲服務

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

下面是一個簡單的實現(xiàn)：

//KafkaDelayQueueProducer延遲隊列生產(chǎn)者，包含了生產(chǎn)者和延遲服務

typeKafkaDelayQueueProducerstruct{

producersarama.SyncProducer//生產(chǎn)者

delayTopicstring//延遲服務主題

//NewKafkaDelayQueueProducer創(chuàng)建延遲隊列生產(chǎn)者

//producer生產(chǎn)者

//delayServiceConsumerGroup延遲服務消費者

//delayTime延遲時間

//delayTopic延遲服務主題

//realTopic真實隊列主題

funcNewKafkaDelayQueueProducer(producersarama.SyncProducer,delayServiceConsumerGroupsarama.ConsumerGroup,

delayTimetime.Duration,delayTopic,realTopicstring)*KafkaDelayQueueProducer{

//啟動延遲服務

consumer:=NewDelayServiceConsumer(producer,delayTime,realTopic)

gofunc(){

for{

iferr:=delayServiceConsumerGroup.Consume(context.Background(),

[]string{delayTopic},consumer);err!=nil{

break

returnKafkaDelayQueueProducer{

producer:producer,

delayTopic:delayTopic,

//SendMessage發(fā)送消息

func(q*KafkaDelayQueueProducer)SendMessage(msg*sarama.ProducerMessage)(partitionint32,offsetint64,errerror){

msg.Topic=q.delayTopic

returnducer.SendMessage(msg)

//DelayServiceConsumer延遲服務消費者

typeDelayServiceConsumerstruct{

producersarama.SyncProducer

delaytime.Duration

realTopicstring

funcNewDelayServiceConsumer(producersarama.SyncProducer,delaytime.Duration,

realTopicstring)*DelayServiceConsumer{

returnDelayServiceConsumer{

producer:producer,

delay:delay,

realTopic:realTopic,

func(c*DelayServiceConsumer)ConsumeClaim(sessionsarama.ConsumerGroupSession,

claimsarama.ConsumerGroupClaim)error{

formessage:=rangeclaim.Messages(){

//如果消息已經(jīng)超時，把消息發(fā)送到真實隊列

now:=time.Now()

ifnow.Sub(message.Timestamp)=c.delay{

_,_,err:=ducer.SendMessage(sarama.ProducerMessage{

Topic:c.realTopic,

Key:sarama.ByteEncoder(message.Key),

Value:sarama.ByteEncoder(message.Value),

iferr==nil{

session.MarkMessage(message,"")

continue

//否則休眠一秒

time.Sleep(time.Second)

returnnil

returnnil

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

returnnil

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