RabbitMq如何做到消息的可靠性投遞
如何保證消息不丟失
在使用RabbitMQ的時候，我們需要保證消息不能丟失，消息從生產(chǎn)者生產(chǎn)出來一直到消費者消費成功，這條鏈路是這樣的：
消息的可靠投遞分為了兩大內(nèi)容：發(fā)送端的確認（p->broker和exchange->queue）和消費端的確認（queue->c）。
發(fā)送端的確認
Rabbit提供了兩種方式來保證發(fā)送端的消息可靠性投遞：confirm 確認模式
和return 退回模式。
confirm 確認模式：消息從 producer 到達 exchange 則會給 producer 發(fā)送一個應(yīng)答，我們需要開啟confirm模式，才能接收到這條應(yīng)答。開啟方式是將Channel.Confirm(noWait bool)參數(shù)設(shè)置為false，表示同意發(fā)送者將當前channel信道設(shè)置為confirm模式。
return 退回模式：消息從 exchange–>queue 投遞失敗，會將消息退回給producer。
消費端的確認
消息從Queue發(fā)送到消費端之后，消費端會發(fā)送一個確認消息：Consumer Ack，有兩種確認方式：自動確認和手動確認。
在編碼中，關(guān)于消息的確認方式，我們需要在消費者端調(diào)用Consumer函數(shù)時，設(shè)置第三個參數(shù)：autoAck是false還是true（false表示手動，true表示自動）。
自動確認是指，當消息一旦被Consumer接收到，則自動確認收到，并將相應(yīng) message 從 RabbitMQ 的消息緩存中移除。
但是在實際業(yè)務(wù)處理中，很可能消息接收到，業(yè)務(wù)處理出現(xiàn)異常，那么該消息就會丟失。如果設(shè)置了手動確認方式，則需要在業(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ā)送端的確認
首先初始化消息隊列的時候，我們要開啟confirm模式，才能接收到這條應(yīng)答。開啟方式是將Channel.Confirm(noWait bool)參數(shù)設(shè)置為false，表示同意發(fā)送者將當前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}
消費端的確認
首先將Consume函數(shù)的第三個參數(shù)autoAck參數(shù)標記為false：
// Consume 接收某個消息隊列的消息
func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
c, e := q.channel.Consume(
q.Name,
"",
false, // 不自動確認消息
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-10
本文主要介紹了Go語言使用buffer讀取文件的實現(xiàn)示例，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-04-04
這篇文章主要介紹了go語言中的return語句，文章圍繞主題展開詳細的內(nèi)容介紹，具有一定的參考價值，需要的小伙伴可以參考一下，希望對你的學(xué)習(xí)有所幫助2022-05-05
這篇文章主要介紹了讓goland支持proto文件類型的實現(xiàn)操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-12-12
本文主要介紹了Go語言zip文件的讀寫操作，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-04-04
在開發(fā)中我們要想編寫高性能的代碼，或者優(yōu)化代碼的性能時，你首先得知道當前代碼的性能，在go中可以使用testing包的benchmark來做基準測試 ，文中有詳細的代碼示例，感興趣的小伙伴可以參考一下2023-04-04
golang結(jié)構(gòu)體與json格式串實例代碼
本文通過實例代碼給大家介紹了golang結(jié)構(gòu)體與json格式串的相關(guān)知識，非常不錯，具有一定的參考借鑒價值,需要的朋友可以參考下2018-10-10
golang中也實現(xiàn)了排序算法的包sort包，下面這篇文章就來給大家介紹了關(guān)于go語言中排序sort的使用方法，文中通過示例代碼介紹的非常詳細，需要的朋友可以參考借鑒，下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2018-06-06
這篇文章主要介紹了Go語言對JSON進行編碼和解碼的方法,涉及Go語言操作json的技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-02-02