关于golang监听rabbitmq消息队列任务断线⾃动重连接的问题 golang监听消息队列rabbitmq任务脚本，当rabbimq消息队列断开连接后⾃动重试，重新唤起协程执⾏任务
需求背景：
goalng常驻内存任务脚本监听rbmq执⾏任务
任务脚本由supervisor来管理
当rabbitmq长时间断开连接会出现如下图进程处于fatal状态
假如因为不可抗拒因素，rabbitmq服务器内存满了或者其它原因导致rabbitmq消息队列服务停⽌了
如果是短时间的停⽌重启，supervisor是可以即时唤醒该程序。如果服务器长时间没有恢复正常运⾏，程序就会出现fatal进程启动失败的状态，此时可以通过告警来提醒开发⼈员 如果以上告警能时时通知运维⼈员此问题可以略过了。今天讨论的是如果在长时间断开连接还能在服务器恢复正常情况下⾃动实现重连。
代码实现⼀：
消费者：
package main
import (
"fmt"
"github/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:guest@192.168.2.232:5672/",
//},t,5)
/*
runNums: 表⽰任务并发处理数量⼀般建议普通任务1-3 就可以了
*/
err := rabbitmq.Recv(rabbitmq.QueueExchange{
"a_test_0001",
"hello_go",
"direct",
"amqp://guest:guest@192.168.1.169:5672/",
},t,4)
if(err != nil){
fmt.Println(err)
}
rabbitmq代码
package rabbitmq
import (
再生油
"errors"
"strconv"
"time"
//"errors"
"fmt"
"github/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 // 交换机类型
// 链接rabbitMQ
func (r *RabbitMQ)MqConnect() (err error){
mqConn, err = amqp.Dial(r.dns)
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断线⾃动重连内容请搜索以前的⽂章或继续浏览下⾯的相关⽂章希望⼤家以后多多⽀持！
豫公网安备41160202000603
站长QQ:729038198
关于我们
投诉建议