关于golang监听rabbitmq消息队列任务断线自动重连接的问题
时间：2022-07-25 10:53:57 | 栏目：Golang | 点击：次
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: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.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)