RabbitMq如何做到消息的可靠性投遞
如何保證消息不丟失
在使用RabbitMQ的時候,我們需要保證消息不能丟失,消息從生產(chǎn)者生產(chǎn)出來一直到消費者消費成功,這條鏈路是這樣的:
消息的可靠投遞分為了兩大內(nèi)容:發(fā)送端的確認(rèn)(p->broker和exchange->queue)和消費端的確認(rèn)(queue->c)。
發(fā)送端的確認(rèn)
Rabbit提供了兩種方式來保證發(fā)送端的消息可靠性投遞:confirm 確認(rèn)模式
和return 退回模式。
confirm 確認(rèn)模式:消息從 producer 到達 exchange 則會給 producer 發(fā)送一個應(yīng)答,我們需要開啟confirm模式,才能接收到這條應(yīng)答。開啟方式是將Channel.Confirm(noWait bool)
參數(shù)設(shè)置為false
,表示同意發(fā)送者將當(dāng)前channel信道設(shè)置為confirm模式。
return 退回模式:消息從 exchange–>queue 投遞失敗,會將消息退回給producer。
消費端的確認(rèn)
消息從Queue發(fā)送到消費端之后,消費端會發(fā)送一個確認(rèn)消息:Consumer Ack
,有兩種確認(rèn)方式:自動確認(rèn)和手動確認(rèn)。
在編碼中,關(guān)于消息的確認(rèn)方式,我們需要在消費者端調(diào)用Consumer
函數(shù)時,設(shè)置第三個參數(shù):autoAck
是false還是true(false表示手動,true表示自動)。
自動確認(rèn)是指,當(dāng)消息一旦被Consumer接收到,則自動確認(rèn)收到,并將相應(yīng) message 從 RabbitMQ 的消息緩存中移除。
但是在實際業(yè)務(wù)處理中,很可能消息接收到,業(yè)務(wù)處理出現(xiàn)異常,那么該消息就會丟失。如果設(shè)置了手動確認(rèn)方式,則需要在業(yè)務(wù)處理成功后,調(diào)用ch.Ack(false)
,手動簽收,如果出現(xiàn)異常,則調(diào)用d.Reject(true)
讓其自動重新發(fā)送消息。
Go 實現(xiàn)
安裝操作庫
安裝API庫
Go可以使用streadway/amqp
庫來操作rabbit,使用以下命令來安裝:
go get github.com/streadway/amqp
封裝rabbitmq
接下來我們對streadway/amqp
庫的內(nèi)容進行一個二次封裝,封裝為一個rabbitmq.go
文件:
package rabbitmq import ( "encoding/json" "github.com/streadway/amqp" "log" ) // RabbitMQ RabbitMQ結(jié)構(gòu) type RabbitMQ struct { channel *amqp.Channel Name string exchange string } // Connect 連接服務(wù)器 func Connect(s string) *RabbitMQ { //連接rabbitmq conn, e := amqp.Dial(s) failOnError(e, "連接Rabbitmq服務(wù)器失??!") ch, e := conn.Channel() failOnError(e, "無法打開頻道!") mq := new(RabbitMQ) mq.channel = ch return mq } // New 初始化消息隊列 //第一個參數(shù):rabbitmq服務(wù)器的鏈接,第二個參數(shù):隊列名字 func New(s string, name string) *RabbitMQ { //連接rabbitmq conn, e := amqp.Dial(s) failOnError(e, "連接Rabbitmq服務(wù)器失??!") ch, e := conn.Channel() failOnError(e, "無法打開頻道!") q, e := ch.QueueDeclare( name, //隊列名 false, //是否開啟持久化 true, //不使用時刪除 false, //排他 false, //不等待 nil, //參數(shù) ) failOnError(e, "初始化消息隊列失??!") mq := new(RabbitMQ) mq.channel = ch mq.Name = q.Name return mq } // QueueDeclare 聲明queue func (q *RabbitMQ) QueueDeclare(queue string) { _, e := q.channel.QueueDeclare(queue, false, true, false, false, nil) failOnError(e, "聲明queue失??!") } // QueueDelete 刪除queue func (q *RabbitMQ) QueueDelete(queue string) { _, e := q.channel.QueueDelete(queue, false, true, false) failOnError(e, "刪除queue失??!") } // Qos 配置queue參數(shù) func (q *RabbitMQ) Qos() { e := q.channel.Qos(1, 0, false) failOnError(e, "無法設(shè)置QoS") } // NewExchange 初始化交換機 //第一個參數(shù):rabbitmq服務(wù)器的鏈接,第二個參數(shù):交換機名字,第三個參數(shù):交換機類型 func NewExchange(s string, name string, typename string) { //連接rabbitmq conn, e := amqp.Dial(s) failOnError(e, "連接Rabbitmq服務(wù)器失??!") ch, e := conn.Channel() failOnError(e, "無法打開頻道!") e = ch.ExchangeDeclare( name, // name typename, // type true, // durable false, // auto-deleted false, // internal false, // no-wait nil, // arguments ) failOnError(e, "初始化交換機失??!") } // ExchangeDelete 刪除交換機 func (q *RabbitMQ) ExchangeDelete(exchange string) { e := q.channel.ExchangeDelete(exchange, false, true) failOnError(e, "刪除交換機失敗!") } // Bind 綁定消息隊列到exchange func (q *RabbitMQ) Bind(exchange string, key string) { e := q.channel.QueueBind( q.Name, key, exchange, false, nil, ) failOnError(e, "綁定隊列失?。?) q.exchange = exchange } // Send 向消息隊列發(fā)送消息 //Send方法可以往某個消息隊列發(fā)送消息 func (q *RabbitMQ) Send(queue string, body interface{}) { str, e := json.Marshal(body) failOnError(e, "消息序列化失?。?) e = q.channel.Publish( "", //交換 queue, //路由鍵 false, //必填 false, //立即 amqp.Publishing{ ReplyTo: q.Name, Body: []byte(str), }) msg := "向隊列:" + q.Name + "發(fā)送消息失?。? failOnError(e, msg) } // Publish 向exchange發(fā)送消息 //Publish方法可以往某個exchange發(fā)送消息 func (q *RabbitMQ) Publish(exchange string, body interface{}, key string) { str, e := json.Marshal(body) failOnError(e, "消息序列化失??!") e = q.channel.Publish( exchange, key, false, false, amqp.Publishing{ReplyTo: q.Name, Body: []byte(str)}, ) failOnError(e, "向交換機發(fā)送消息失敗!") } // Consume 接收某個消息隊列的消息 func (q *RabbitMQ) Consume() <-chan amqp.Delivery { c, e := q.channel.Consume( q.Name, //指定從哪個隊列中接收消息 "", true, false, false, false, nil, ) failOnError(e, "接收消息失敗!") return c } // Close 關(guān)閉隊列連接 func (q *RabbitMQ) Close() { q.channel.Close() } //錯誤處理函數(shù) func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) } }
發(fā)送端的確認(rèn)
首先初始化消息隊列的時候,我們要開啟confirm模式
,才能接收到這條應(yīng)答。開啟方式是將Channel.Confirm(noWait bool)
參數(shù)設(shè)置為false
,表示同意發(fā)送者將當(dāng)前channel信道設(shè)置為confirm模式。
func New(s string, name string) *RabbitMQ { conn, e := amqp.Dial(s) failOnError(e, "連接Rabbitmq服務(wù)器失?。?) ch, e := conn.Channel() failOnError(e, "無法打開頻道!") q, e := ch.QueueDeclare( name, //隊列名 false, //是否開啟持久化 true, //不使用時刪除 false, //排他 false, //不等待 nil, //參數(shù) ) failOnError(e, "初始化消息隊列失敗!") mq := new(RabbitMQ) mq.channel = ch mq.Name = q.Name // 設(shè)置為confirm模式 mq.channel.Confirm(false) return mq }
然后在封裝庫中創(chuàng)建一個函數(shù)handleConfirm()
用于接收來自Borker的回復(fù):
func (q *RabbitMQ) ConfirmFromBroker(ch chan amqp.Confirmation) chan amqp.Confirmation { return q.channel.NotifyPublish(ch) }
生產(chǎn)者
生產(chǎn)者端在向Broker發(fā)送消息的時候,我們使用一個無緩沖的通道來接收來自Broker的回復(fù),然后創(chuàng)建一個協(xié)程監(jiān)聽這個無緩沖通道。
func main() { producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue") // 指定為topic類型 rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange1", "fanout") confirm := producer.ConfirmFromBroker(make(chan amqp.Confirmation)) go handleConfirm(confirm) var i int for { time.Sleep(time.Second) producer.Publish("exchange1", "fanout message: "+strconv.Itoa(i), "") i++ } } func handleConfirm(confirm <-chan amqp.Confirmation) { for { select { case message := <-confirm: fmt.Println("接收到來自Broker的回復(fù):", message) } } }
運行結(jié)果:
接收到來自Broker的回復(fù): {1 true}
接收到來自Broker的回復(fù): {2 true}
接收到來自Broker的回復(fù): {3 true}
接收到來自Broker的回復(fù): {4 true}
接收到來自Broker的回復(fù): {5 true}
消費端的確認(rèn)
首先將Consume
函數(shù)的第三個參數(shù)autoAck
參數(shù)標(biāo)記為false:
// Consume 接收某個消息隊列的消息 func (q *RabbitMQ) Consume() <-chan amqp.Delivery { c, e := q.channel.Consume( q.Name, "", false, // 不自動確認(rèn)消息 false, false, false, nil, ) failOnError(e, "接收消息失敗!") return c }
在消費者端我們采用公平派遣模式,即隊列發(fā)送消息給消費者的時候,不再采用輪詢機制,而是一個消費者消費完消息之后,會調(diào)用Ack(false)
函數(shù)向隊列發(fā)送一個回復(fù),隊列每次會將消息優(yōu)先發(fā)送給消費完消息的消費者(回復(fù)過)。
消費端限流:
實現(xiàn)公平派遣模式我們需要設(shè)置消費者端一次只能消費一條消息,之前我們已經(jīng)進行了封裝,直接在消費者端調(diào)用即可:
// Qos 配置queue參數(shù) func (q *RabbitMQ) Qos() { e := q.channel.Qos(1, 0, false) failOnError(e, "無法設(shè)置QoS") }
生產(chǎn)者
func main() { producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue") // 指定為direct類型 rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange", "direct") i := 0 for { time.Sleep(time.Second) producer.Publish("exchange", "routing message: "+strconv.Itoa(i), "key1") i = i + 1 } }
消費者1
消費者2在消費第三條消息的時候,假設(shè)發(fā)生了錯誤,我們調(diào)用d.Reject(true)
函數(shù)讓隊列重新發(fā)送消息。
func main() { //第一個參數(shù)指定rabbitmq服務(wù)器的鏈接,第二個參數(shù)指定創(chuàng)建隊列的名字 consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1") // 指定一次只消費一條消息,直到消費完才重新接收 consumer1.Qos() // 隊列綁定到exchange consumer1.Bind("exchange", "key1") //接收消息 msgs := consumer1.Consume() go func() { var i int for d := range msgs { time.Sleep(time.Second * 1) log.Printf("Consumer1 received a message: %s", d.Body) // 假設(shè)消費第三條消息的時候出現(xiàn)了錯誤,我們就調(diào)用d.Reject(true),隊列會重新發(fā)送消息給消費者 if i == 2 { d.Reject(true) } else { // 消息消費成功之后就回復(fù) d.Ack(false) } i++ } }() select {} }
消費者2
func main() { //第一個參數(shù)指定rabbitmq服務(wù)器的鏈接,第二個參數(shù)指定創(chuàng)建隊列的名字 consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1") // 指定一次只消費一條消息,直到消費完才重新接收 consumer2.Qos() // 隊列綁定到exchange consumer2.Bind("exchange", "key1") //接收消息 msgs := consumer2.Consume() go func() { for d := range msgs { time.Sleep(time.Second * 5) log.Printf("Consumer2 received a message: %s", d.Body) // 消息消費成功之后就回復(fù) d.Ack(false) } }() select {} }
運行結(jié)果:
# 消費者1
2022/11/06 19:55:08 Consumer1 received a message: "routing message: 0"
2022/11/06 19:55:10 Consumer1 received a message: "routing message: 2"
2022/11/06 19:55:11 Consumer1 received a message: "routing message: 3"
2022/11/06 19:55:12 Consumer1 received a message: "routing message: 3"
2022/11/06 19:55:13 Consumer1 received a message: "routing message: 4"
2022/11/06 19:55:14 Consumer1 received a message: "routing message: 6"# 消費者2
2022/11/06 19:55:13 Consumer2 received a message: "routing message: 1"
到此這篇關(guān)于RabbitMq如何做到消息的可靠性投遞的文章就介紹到這了,更多相關(guān)RabbitMq消息可靠性投遞內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Go并發(fā)編程中的錯誤恢復(fù)機制與代碼持續(xù)執(zhí)行實例探索
這篇文章主要為大家介紹了Go并發(fā)編程中的錯誤恢復(fù)機制與代碼持續(xù)執(zhí)行實例探索,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2024-01-01數(shù)據(jù)競爭和內(nèi)存重分配Golang slice并發(fā)不安全問題解決
這篇文章主要為大家介紹了數(shù)據(jù)競爭和內(nèi)存重分配Golang slice并發(fā)不安全問題解決,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-10-10golang結(jié)構(gòu)體與json格式串實例代碼
本文通過實例代碼給大家介紹了golang結(jié)構(gòu)體與json格式串的相關(guān)知識,非常不錯,具有一定的參考借鑒價值,需要的朋友可以參考下2018-10-10