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

第Go+Kafka實(shí)現(xiàn)延遲消息的實(shí)現(xiàn)示例目錄前言原理簡單的實(shí)現(xiàn)生產(chǎn)者延遲服務(wù)消費(fèi)者改進(jìn)點(diǎn)通用的延遲服務(wù)生產(chǎn)者負(fù)責(zé)延遲服務(wù)總結(jié)

前言

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

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

使用了sarama客戶端。

原理

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

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

簡單的實(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ù)會訂閱延遲隊(duì)列的消息，并把超時消息發(fā)送到真實(shí)隊(duì)列

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ā)送到真實(shí)隊(duì)列

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

}

消費(fèi)者

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

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

}

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

通用的延遲服務(wù)

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

延遲服務(wù)可以提供多個延時等級，比如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ì)列。

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

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

typeKafkaDelayQueueProducerstruct{

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

delayTopicstring//延遲服務(wù)主題

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

//producer生產(chǎn)者

//delayServiceConsumerGroup延遲服務(wù)消費(fèi)者

//delayTime延遲時間

//delayTopic延遲服務(wù)主題

//realTopic真實(shí)隊(duì)列主題

funcNewKafkaDelayQueueProducer(producersarama.SyncProducer,delayServiceConsumerGroupsarama.ConsumerGroup,

delayTimetime.Duration,delayTopic,realTopicstring)*KafkaDelayQueueProducer{

//啟動延遲服務(wù)

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延遲服務(wù)消費(fèi)者

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ā)送到真實(shí)隊(duì)列

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{