關於golang監聽rabbitmq消息隊列任務斷線自動重連接的問題
golang監聽消息隊列rabbitmq任務腳本，當rabbimq消息隊列斷開連接後自動重試，重新喚起協程執行任務
需求背景：
goalng常駐內存任務腳本監聽rbmq執行任務
任務腳本由supervisor來管理
當rabbitmq長時間斷開連接會出現如下圖 進程處於fatal狀態
假如因為不可抗拒因素，rabbitmq服務器內存滿瞭或者其它原因導致rabbitmq消息隊列服務停止瞭
如果是短時間的停止重啟，supervisor是可以即時喚醒該程序。如果服務器長時間沒有恢復正常運行，程序就會出現fatal進程啟動失敗的狀態，此時可以通過告警來提醒開發人員
如果以上告警能時時通知運維人員此問題可以略過瞭。今天討論的是如果在長時間斷開連接還能在服務器恢復正常情況下自動實現重連。
代碼實現一：
消費者：
package main
import (
"fmt"
"github.com/ichunt2019/golang-rbmq-sl/utils/rabbitmq"
)
type RecvPro struct {
}
//// 實現消費者 消費消息失敗 自動進入延時嘗試 嘗試3次之後入庫db
/*
返回值 error 為nil 則表示該消息消費成功
否則消息會進入ttl延時隊列 重復嘗試消費3次
3次後消息如果還是失敗 消息就執行失敗 進入告警 FailAction
*/
func (t *RecvPro) Consumer(dataByte []byte) error {
//time.Sleep(500*time.Microsecond)
//return errors.New("頂頂頂頂")
fmt.Println(string(dataByte))
//time.Sleep(1*time.Second)
return nil
//消息已經消費3次 失敗瞭 請進行處理
如果消息 消費3次後 仍然失敗 此處可以根據情況 對消息進行告警提醒 或者 補償 入庫db 釘釘告警等等
func (t *RecvPro) FailAction(err error,dataByte []byte) error {
fmt.Println(err)
fmt.Println("任務處理失敗瞭，我要進入db日志庫瞭")
fmt.Println("任務處理失敗瞭，發送釘釘消息通知主人")
func main() {
t := &RecvPro{}
//rabbitmq.Recv(rabbitmq.QueueExchange{
// "a_test_0001",
// "",
// "amqp://guest:[email protected]:5672/",
//},t,5)
/*
runNums: 表示任務並發處理數量 一般建議 普通任務1-3 就可以瞭
*/
err := rabbitmq.Recv(rabbitmq.QueueExchange{
"a_test_0001",
"hello_go",
"direct",
"amqp://guest:[email protected]:5672/",
},t,4)
if(err != nil){
fmt.Println(err)
}
rabbitmq代碼
package rabbitmq
import (
"errors"
"strconv"
"time"
//"errors"
"fmt"
"github.com/streadway/amqp"
"log"
)
// 定義全局變量,指針類型
var mqConn *amqp.Connection
var mqChan *amqp.Channel
// 定義生產者接口
type Producer interface {
MsgContent() string
}
type RetryProducer interface {
// 定義接收者接口
type Receiver interface {
Consumer([]byte) error
FailAction(error , []byte) error
// 定義RabbitMQ對象
type RabbitMQ struct {
connection *amqp.Connection
Channel *amqp.Channel
dns string
QueueName string // 隊列名稱
RoutingKey string // key名稱
ExchangeName string // 交換機名稱
ExchangeType string // 交換機類型
producerList []Producer
retryProducerList []RetryProducer
receiverList []Receiver
// 定義隊列交換機對象
type QueueExchange struct {
QuName string // 隊列名稱
RtKey string // key值
ExName string // 交換機名稱
ExType string // 交換機類型
Dns string //鏈接地址
// 鏈接rabbitMQ
func (r *RabbitMQ)MqConnect() (err error){
mqConn, err = amqp.Dial(r.dns)
r.connection = mqConn // 賦值給RabbitMQ對象
if err != nil {
fmt.Printf("rbmq鏈接失敗 :%s \n", err)
}
return
// 關閉mq鏈接
func (r *RabbitMQ)CloseMqConnect() (err error){
err = r.connection.Close()
if err != nil{
fmt.Printf("關閉mq鏈接失敗 :%s \n", err)
func (r *RabbitMQ)MqOpenChannel() (err error){
mqConn := r.connection
r.Channel, err = mqConn.Channel()
//defer mqChan.Close()
fmt.Printf("MQ打開管道失敗:%s \n", err)
return err
func (r *RabbitMQ)CloseMqChannel() (err error){
r.Channel.Close()
// 創建一個新的操作對象
func NewMq(q QueueExchange) RabbitMQ {
return RabbitMQ{
QueueName:q.QuName,
RoutingKey:q.RtKey,
ExchangeName: q.ExName,
ExchangeType: q.ExType,
dns:q.Dns,
func (mq *RabbitMQ) sendMsg (body string) (err error) {
err = mq.MqOpenChannel()
ch := mq.Channel
log.Printf("Channel err :%s \n", err)
defer mq.Channel.Close()
if mq.ExchangeName != "" {
if mq.ExchangeType == ""{
mq.ExchangeType = "direct"
}
err = ch.ExchangeDeclare(mq.ExchangeName, mq.ExchangeType, true, false, false, false, nil)
if err != nil {
log.Printf("ExchangeDeclare err :%s \n", err)
// 用於檢查隊列是否存在,已經存在不需要重復聲明
_, err = ch.QueueDeclare(mq.QueueName, true, false, false, false, nil)
log.Printf("QueueDeclare err :%s \n", err)
// 綁定任務
if mq.RoutingKey != "" && mq.ExchangeName != "" {
err = ch.QueueBind(mq.QueueName, mq.RoutingKey, mq.ExchangeName, false, nil)
log.Printf("QueueBind err :%s \n", err)
if mq.ExchangeName != "" && mq.RoutingKey != ""{
err = mq.Channel.Publish(
mq.ExchangeName, // exchange
mq.RoutingKey, // routing key
false, // mandatory
false, // immediate
amqp.Publishing {
ContentType: "text/plain",
Body: []byte(body),
})
}else{
"", // exchange
mq.QueueName, // routing key
/*
發送延時消息
*/
func (mq *RabbitMQ)sendDelayMsg(body string,ttl int64) (err error){
err =mq.MqOpenChannel()
return
if ttl <= 0{
return errors.New("發送延時消息，ttl參數是必須的")
table := make(map[string]interface{},3)
table["x-dead-letter-routing-key"] = mq.RoutingKey
table["x-dead-letter-exchange"] = mq.ExchangeName
table["x-message-ttl"] = ttl*1000
//fmt.Printf("%+v",table)
//fmt.Printf("%+v",mq)
ttlstring := strconv.FormatInt(ttl,10)
queueName := fmt.Sprintf("%s_delay_%s",mq.QueueName ,ttlstring)
routingKey := fmt.Sprintf("%s_delay_%s",mq.QueueName ,ttlstring)
_, err = ch.QueueDeclare(queueName, true, false, false, false, table)
return
if routingKey != "" && mq.ExchangeName != "" {
err = ch.QueueBind(queueName, routingKey, mq.ExchangeName, false, nil)
header := make(map[string]interface{},1)
header["retry_nums"] = 0
var ttl_exchange string
var ttl_routkey string
if(mq.ExchangeName != "" ){
ttl_exchange = mq.ExchangeName
ttl_exchange = ""
if mq.RoutingKey != "" && mq.ExchangeName != ""{
ttl_routkey = routingKey
ttl_routkey = queueName
err = mq.Channel.Publish(
ttl_exchange, // exchange
ttl_routkey, // routing key
false, // mandatory
false, // immediate
amqp.Publishing {
ContentType: "text/plain",
Body: []byte(body),
Headers:header,
})
func (mq *RabbitMQ) sendRetryMsg (body string,retry_nums int32,args ...string) {
err :=mq.MqOpenChannel()
//原始路由key
oldRoutingKey := args[0]
//原始交換機名
oldExchangeName := args[1]
table["x-dead-letter-routing-key"] = oldRoutingKey
if oldExchangeName != "" {
table["x-dead-letter-exchange"] = oldExchangeName
mq.ExchangeName = ""
table["x-dead-letter-exchange"] = ""
table["x-message-ttl"] = int64(20000)
_, err = ch.QueueDeclare(mq.QueueName, true, false, false, false, table)
header["retry_nums"] = retry_nums + int32(1)
ttl_routkey = mq.RoutingKey
ttl_routkey = mq.QueueName
//fmt.Printf("ttl_exchange:%s,ttl_routkey:%s \n",ttl_exchange,ttl_routkey)
fmt.Printf("MQ任務發送失敗:%s \n", err)
// 監聽接收者接收任務 消費者
func (mq *RabbitMQ) ListenReceiver(receiver Receiver) {
// 獲取消費通道,確保rabbitMQ一個一個發送消息
err = ch.Qos(1, 0, false)
msgList, err := ch.Consume(mq.QueueName, "", false, false, false, false, nil)
log.Printf("Consume err :%s \n", err)
for msg := range msgList {
retry_nums,ok := msg.Headers["retry_nums"].(int32)
if(!ok){
retry_nums = int32(0)
// 處理數據
err := receiver.Consumer(msg.Body)
if err!=nil {
//消息處理失敗 進入延時嘗試機制
if retry_nums < 3{
fmt.Println(string(msg.Body))
fmt.Printf("消息處理失敗 消息開始進入嘗試 ttl延時隊列 \n")
retry_msg(msg.Body,retry_nums,QueueExchange{
mq.QueueName,
mq.RoutingKey,
mq.ExchangeName,
mq.ExchangeType,
mq.dns,
})
}else{
//消息失敗 入庫db
fmt.Printf("消息處理3次後還是失敗瞭 入庫db 釘釘告警 \n")
receiver.FailAction(err,msg.Body)
}
err = msg.Ack(true)
if err != nil {
fmt.Printf("確認消息未完成異常:%s \n", err)
}else {
// 確認消息,必須為false
fmt.Printf("消息消費ack失敗 err :%s \n", err)
//消息處理失敗之後 延時嘗試
func retry_msg(msg []byte,retry_nums int32,queueExchange QueueExchange){
//原始隊列名稱 交換機名稱
oldQName := queueExchange.QuName
oldExchangeName := queueExchange.ExName
oldRoutingKey := queueExchange.RtKey
if oldRoutingKey == "" || oldExchangeName == ""{
oldRoutingKey = oldQName
if queueExchange.QuName != "" {
queueExchange.QuName = queueExchange.QuName + "_retry_3";
if queueExchange.RtKey != "" {
queueExchange.RtKey = queueExchange.RtKey + "_retry_3";
queueExchange.RtKey = queueExchange.QuName + "_retry_3";
//fmt.Printf("%+v",queueExchange)
mq := NewMq(queueExchange)
_ = mq.MqConnect()
defer func(){
_ = mq.CloseMqConnect()
}()
//fmt.Printf("%+v",queueExchange)
mq.sendRetryMsg(string(msg),retry_nums,oldRoutingKey,oldExchangeName)
func Send(queueExchange QueueExchange,msg string) (err error){
err = mq.MqConnect()
mq.CloseMqConnect()
err = mq.sendMsg(msg)
//發送延時消息
func SendDelay(queueExchange QueueExchange,msg string,ttl int64)(err error){
err = mq.sendDelayMsg(msg,ttl)
runNums 開啟並發執行任務數量
func Recv(queueExchange QueueExchange,receiver Receiver,runNums int) (err error){
//鏈接rabbitMQ
if(err != nil){
//rbmq斷開鏈接後 協程退出釋放信號
taskQuit:= make(chan struct{}, 1)
//嘗試鏈接rbmq
tryToLinkC := make(chan struct{}, 1)
//開始執行任務
for i:=1;i<=runNums;i++{
go Recv2(mq,receiver,taskQuit);
//如果rbmq斷開連接後 嘗試重新建立鏈接
var tryToLink = func() {
for {
err = mq.MqConnect()
if(err == nil){
tryToLinkC <- struct{}{}
break
time.Sleep(time.Second * 10)
for{
select {
case <- taskQuit ://rbmq斷開連接後 開始嘗試重新建立鏈接
go tryToLink()
<-tryToLinkC //建立鏈接成功後 重新開啟協程執行任務
fmt.Println("重新開啟新的協程執行任務")
go Recv2(mq,receiver,taskQuit);
time.Sleep(time.Millisecond*100)
func Recv2(mq RabbitMQ,receiver Receiver,taskQuit chan<- struct{}){
defer func() {
fmt.Println("rbmq鏈接失敗,協程任務退出~~~~~~~~~~~~~~~~~~~~")
taskQuit <- struct{}{}
}()
// 驗證鏈接是否正常
err := mq.MqOpenChannel()
if(err != nil){
mq.ListenReceiver(receiver)
type retryPro struct {
msgContent string
實現重連方式很多，下面實現方式比較簡單
1.Recv方法創建ampq鏈接
2.啟動協程開始執行任務
MqOpenChannel 打開一個channel通道處理amqp消息
拿到消息 處理任務
3，協程中捕獲異常發送消息到taskQuit <- struct{}{}
4，主進程監聽taskQuit管道 開始嘗試重新鏈接amqp 直到鏈接成功
5，重新鏈接成功後啟動新的協程處理任務
主要代碼分析：
/*
runNums 開啟並發執行任務數量
*/
func Recv(queueExchange QueueExchange,receiver Receiver,runNums int) (err error){
mq := NewMq(queueExchange)
//鏈接rabbitMQ
err = mq.MqConnect()
if(err != nil){
return
}
//rbmq斷開鏈接後 協程退出釋放信號
taskQuit:= make(chan struct{}, 1)
//嘗試鏈接rbmq
tryToLinkC := make(chan struct{}, 1)
//開始執行任務
for i:=1;i<=runNums;i++{
go Recv2(mq,receiver,taskQuit);
//如果rbmq斷開連接後 嘗試重新建立鏈接
var tryToLink = func() {
for {
err = mq.MqConnect()
if(err == nil){
tryToLinkC <- struct{}{}
break
}
time.Sleep(time.Second * 10)
}
for{
select {
case <- taskQuit ://rbmq斷開連接後 開始嘗試重新建立鏈接
go tryToLink()
<-tryToLinkC //建立鏈接成功後 重新開啟協程執行任務
fmt.Println("重新開啟新的協程執行任務")
go Recv2(mq,receiver,taskQuit);
time.Sleep(time.Millisecond*100)
}
func Recv2(mq RabbitMQ,receiver Receiver,taskQuit chan<- struct{}){
defer func() {
fmt.Println("rbmq鏈接失敗,協程任務退出~~~~~~~~~~~~~~~~~~~~")
taskQuit <- struct{}{}
return
}()
// 驗證鏈接是否正常
err := mq.MqOpenChannel()
if(err != nil){
mq.ListenReceiver(receiver)
到此這篇關於golang監聽rabbitmq消息隊列任務斷線自動重連接的文章就介紹到這瞭,更多相關golang rabbitmq斷線自動重連內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- C#通過rabbitmq實現定時任務(延時隊列)
- .Net Core和RabbitMQ限制循環消費的方法
- RabbitMQ 延遲隊列實現訂單支付結果異步階梯性通知(實例代碼)
- rabbitmq中routingkey的作用說明
- .net平臺的rabbitmq使用封裝demo詳解