当前位置 : 主页 > 编程语言 > 其它开发 >

golang监听rabbitmq消息队列任务断线自动重连接

来源：互联网 收集：自由互联 发布时间：2022-05-22
需求背景： goalng常驻内存任务脚本监听rbmq执行任务 任务脚本由supervisor来管理 当rabbitmq长时间断开连接会出现如下图 进程处于fatal状态 假如因为不可抗拒因素，rabbitmq服务器内存满了或

需求背景：

goalng常驻内存任务脚本监听rbmq执行任务

任务脚本由supervisor来管理

  当rabbitmq长时间断开连接会出现如下图 进程处于fatal状态

 

 假如因为不可抗拒因素，rabbitmq服务器内存满了或者其它原因导致rabbitmq消息队列服务停止了

如果是短时间的停止重启，supervisor是可以即时唤醒该程序。如果服务器长时间没有恢复正常运行，程序就会出现fatal进程启动失败的状态，此时可以通过告警来提醒开发人员

 

 如果以上告警能时时通知运维人员此问题可以略过了。今天讨论的是如果在长时间断开连接还能在服务器恢复正常情况下自动实现重连。

 

 

实现重连方式很多，下面实现方式比较简单

使用通道实现通知

通知可以被看作是特殊的请求/回应用例。在一个通知用例中，我们并不关心回应的值，我们只关心回应是否已发生。 所以我们常常使用空结构体类型struct{}来做为通道的元素类型，因为空结构体类型的尺寸为零，能够节省一些内存（虽然常常很少量）。

向一个通道发送一个值来实现单对单通知

我们已知道，如果一个通道中无值可接收，则此通道上的下一个接收操作将阻塞到另一个协程发送一个值到此通道为止。 所以一个协程可以向此通道发送一个值来通知另一个等待着从此通道接收数据的协程。

 

  1. Recv方法创建ampq链接
  2. 启动协程开始执行任务
    1. 　　MqOpenChannel 打开一个channel通道处理amqp消息
    2.        拿到消息 处理任务

 　　3，协程中捕获异常发送消息到 taskQuit <- struct{}{}

　　4，主进程监听taskQuit管道 开始尝试重新链接amqp

　　　　4.1如果在规定时间内链接成功  重新链接成功后启动新的协程处理任务

　　 　   4.2如果在规定时间内没有链接成功   则退出主进程

send.go 生产者

package main

import (
    "fmt"
    _ "fmt"
    "github.com/ichunt2019/golang-rabbitmq-2022/utils/rabbitmq"
)

func main() {


    for i := 0;i<20;i++{
        body := fmt.Sprintf("{\"order_id\":%d}",i)
        fmt.Println(body)

        /**
            使用默认的交换机
            如果是默认交换机
            type QueueExchange struct {
            QuName  string           // 队列名称
            RtKey   string           // key值
            ExName  string           // 交换机名称
            ExType  string           // 交换机类型
            Dns     string              //链接地址
            }
            如果你喜欢使用默认交换机
            RtKey  此处建议填写成 RtKey 和 QuName 一样的值
         */

        queueExchange := rabbitmq.QueueExchange{
            "a_test_0001",
            "a_test_0001",
            "hello_go",
            "direct",
            "amqp://guest:guest@192.168.1.169:5672/",
        }

        _ = rabbitmq.Send(queueExchange,body)

    }
}
View Code

 

 

recv.go 消费者

package main

import (
    "fmt"
    "github.com/ichunt2019/golang-rabbitmq-2022/utils/rabbitmq"
    "time"
)

type RecvPro struct {

}

//// 实现消费者 消费消息失败 自动进入延时尝试  尝试3次之后入库db
/*
返回值 error 为nil  则表示该消息消费成功
否则消息会进入ttl延时队列  重复尝试消费3次
3次后消息如果还是失败 消息就执行失败  进入告警 FailAction
 */
func (t *RecvPro) Consumer(dataByte []byte) error {
    time.Sleep(time.Second*1)
    //return errors.New("顶顶顶顶")
    fmt.Println(string(dataByte))
    //time.Sleep(1*time.Second)
    //return errors.New("顶顶顶顶")
    return nil
}

//消息已经消费3次 失败了 请进行处理
/*
如果消息 消费3次后 仍然失败  此处可以根据情况 对消息进行告警提醒 或者 补偿  入库db  钉钉告警等等
 */
func (t *RecvPro) FailAction(err error,dataByte []byte) error {
    fmt.Println(string(dataByte))
    fmt.Println(err)
    fmt.Println("任务处理失败了，我要进入db日志库了")
    fmt.Println("任务处理失败了，发送钉钉消息通知主人")
    return nil
}



func main() {
    processTask := &RecvPro{}

    /*
        runNums: 表示任务并发处理数量  一般建议 普通任务1-3    就可以了
        maxTryConnTimeFromMinute:表示最大尝试时间  分钟
     */
    err := rabbitmq.Recv(rabbitmq.QueueExchange{
        "a_test_0001",
        "a_test_0001",
        "hello_go",
        "direct",
        "amqp://guest:guest@192.168.1.169:5672/",
    },
    processTask,4,2)
    if(err != nil){
        fmt.Println(err)
    }

}
View Code

 

 

receiver.go   golang-rbmq工具包

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 {
    MsgContent() string
}

// 定义接收者接口
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)
    }
    return
}

// 链接rabbitMQ
func (r *RabbitMQ)MqOpenChannel() (err error){
    mqConn := r.connection
    r.Channel, err = mqConn.Channel()
    //defer mqChan.Close()
    if err != nil {
        fmt.Printf("MQ打开管道失败:%s \n", err)
    }
    return err
}

// 链接rabbitMQ
func (r *RabbitMQ)CloseMqChannel() (err error){
    r.Channel.Close()
    if err != nil {
        fmt.Printf("关闭mq链接失败  :%s \n", err)
    }
    return err
}




// 创建一个新的操作对象
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
    if err != nil{
        log.Printf("Channel err  :%s \n", err)
    }

    defer func() {
        _ = 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)
    if err != nil {
        log.Printf("QueueDeclare err :%s \n", err)
    }
    // 绑定任务
    if mq.RoutingKey != "" && mq.ExchangeName != "" {
        err = ch.QueueBind(mq.QueueName, mq.RoutingKey, mq.ExchangeName, false, nil)
        if err != 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),
                DeliveryMode: 2,
            })
    }else{
        err = mq.Channel.Publish(
            "",     // exchange
            mq.QueueName, // routing key
            false,  // mandatory
            false,  // immediate
            amqp.Publishing {
                ContentType: "text/plain",
                Body:        []byte(body),
                DeliveryMode: 2,
            })
    }
    return

}


/*
发送延时消息
 */
func (mq *RabbitMQ)sendDelayMsg(body string,ttl int64) (err error){
    err =mq.MqOpenChannel()
    ch := mq.Channel
    if err != nil{
        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 {
            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)
    if err != nil {
        return
    }
    // 绑定任务
    if routingKey != "" && mq.ExchangeName != "" {
        err = ch.QueueBind(queueName, routingKey, mq.ExchangeName, false, nil)
        if err != nil {
            return
        }
    }

    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
    }else{
        ttl_exchange = ""
    }


    if mq.RoutingKey != "" && mq.ExchangeName != ""{
        ttl_routkey = routingKey
    }else{
        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,
        })
    if err != nil {
        return

    }
    return
}


func (mq *RabbitMQ) sendRetryMsg (body string,retry_nums int32,args ...string)  {
    err :=mq.MqOpenChannel()
    ch := mq.Channel
    if err != nil{
        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)
        }
    }

    //原始路由key
    oldRoutingKey := args[0]
    //原始交换机名
    oldExchangeName := args[1]

    table := make(map[string]interface{},3)
    table["x-dead-letter-routing-key"] = oldRoutingKey
    if oldExchangeName != "" {
        table["x-dead-letter-exchange"] = oldExchangeName
    }else{
        mq.ExchangeName = ""
        table["x-dead-letter-exchange"] = ""
    }

    table["x-message-ttl"] = int64(20000)

    //fmt.Printf("%+v",table)
    //fmt.Printf("%+v",mq)
    // 用于检查队列是否存在,已经存在不需要重复声明
    _, err = ch.QueueDeclare(mq.QueueName, true, false, false, false, table)
    if err != nil {
        log.Printf("QueueDeclare err :%s \n", err)
    }
    // 绑定任务
    if mq.RoutingKey != "" && mq.ExchangeName != "" {
        err = ch.QueueBind(mq.QueueName, mq.RoutingKey, mq.ExchangeName, false, nil)
        if err != nil {
            log.Printf("QueueBind err :%s \n", err)
        }
    }

    header := make(map[string]interface{},1)

    header["retry_nums"] = retry_nums + int32(1)

    var ttl_exchange string
    var ttl_routkey string

    if(mq.ExchangeName != "" ){
        ttl_exchange = mq.ExchangeName
    }else{
        ttl_exchange = ""
    }


    if mq.RoutingKey != "" && mq.ExchangeName != ""{
        ttl_routkey = mq.RoutingKey
    }else{
        ttl_routkey = mq.QueueName
    }

    //fmt.Printf("ttl_exchange:%s,ttl_routkey:%s \n",ttl_exchange,ttl_routkey)
    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,
        })
    if err != nil {
        fmt.Printf("MQ任务发送失败:%s \n", err)

    }

}


// 监听接收者接收任务 消费者
func (mq *RabbitMQ) ListenReceiver(receiver Receiver) {
    err :=mq.MqOpenChannel()
    ch := mq.Channel
    if err != nil{
        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)
    if err != nil {
        log.Printf("QueueDeclare err :%s \n", err)
    }
    // 绑定任务
    if mq.RoutingKey != "" && mq.ExchangeName != "" {
        err = ch.QueueBind(mq.QueueName, mq.RoutingKey, mq.ExchangeName, false, nil)
        if err != nil {
            log.Printf("QueueBind err :%s \n", err)
        }
    }
    // 获取消费通道,确保rabbitMQ一个一个发送消息
    err =  ch.Qos(1, 0, false)
    msgList, err :=  ch.Consume(mq.QueueName, "", false, false, false, false, nil)
    if err != 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
            err = msg.Ack(true)

            if err != nil {
                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";
    }else{
        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){
    mq := NewMq(queueExchange)
    err = mq.MqConnect()
    if err != nil{
        return
    }

    defer func(){
        _ = mq.CloseMqConnect()
    }()

    err = mq.sendMsg(msg)

    return
}

//发送延时消息
func SendDelay(queueExchange QueueExchange,msg string,ttl int64)(err error){
    mq := NewMq(queueExchange)
    err = mq.MqConnect()
    if err != nil{
        return
    }
    defer func(){
        _ = mq.CloseMqConnect()
    }()
    err = mq.sendDelayMsg(msg,ttl)
    return
}


/*
runNums  开启并发执行任务数量
 */
func Recv(queueExchange QueueExchange,receiver Receiver,otherParams ...int) (err error){
    var (
        exitTask bool
        maxTryConnNums int  //rbmq链接失败后多久尝试一次
        runNums int
        maxTryConnTimeFromMinute int
    )

    if(len(otherParams) <= 0){
        runNums = 1
        maxTryConnTimeFromMinute = 0
    }else if(len(otherParams) == 1){
        runNums = otherParams[0]
        maxTryConnTimeFromMinute = 0
    }else if(len(otherParams) == 2){
        runNums = otherParams[0]
        maxTryConnTimeFromMinute = otherParams[1]
    }


    //maxTryConnNums := 360 //rbmq链接失败后最大尝试次数
    //maxTryConnTime := time.Duration(10) //rbmq链接失败后多久尝试一次
    maxTryConnNums = maxTryConnTimeFromMinute * 10 * maxTryConnTimeFromMinute//rbmq链接失败后最大尝试次数
    maxTryConnTime := time.Duration(6) //rbmq链接失败后多久尝试一次
    mq := NewMq(queueExchange)
    //链接rabbitMQ
    err = mq.MqConnect()
    if(err != nil){
        return
    }

    defer func() {
        if panicErr := recover(); panicErr != nil{
            fmt.Println(recover())
            err = errors.New(fmt.Sprintf("%s",panicErr))
        }
    }()

    //rbmq断开链接后 协程退出释放信号
    taskQuit:= make(chan struct{}, 1)
    //尝试链接rbmq
    tryToLinkC := make(chan struct{}, 1)

    //最大尝试次数
    tryToLinkMaxNums := make(chan struct{}, 1)

    maxTryNums := 0 //尝试重启次数

    //开始执行任务
    for i:=1;i<=runNums;i++{
        go Recv2(mq,receiver,taskQuit);
    }

    //如果rbmq断开连接后 尝试重新建立链接
    var tryToLink = func() {
        for {
            maxTryNums += 1
            err = mq.MqConnect()
            if(err == nil){
                tryToLinkC <- struct{}{}
                break
            }
            if(maxTryNums > maxTryConnNums){
                tryToLinkMaxNums <- struct{}{}
                break
            }
            //如果链接断开了 10秒重新尝试链接一次
            time.Sleep(time.Second * maxTryConnTime)
        }
        return
    }
    scheduleTimer := time.NewTimer(time.Millisecond*300)
    exitTask = true
    for{
        select {
        case <-tryToLinkC: //建立链接成功后 重新开启协程执行任务
            fmt.Println("重新开启新的协程执行任务")
            go Recv2(mq,receiver,taskQuit);
        case <-tryToLinkMaxNums://rbmq超出最大链接次数 退出任务
            fmt.Println("rbmq链接超过最大尝试次数!")
            exitTask = false
            err = errors.New("rbmq链接超过最大尝试次数!")
        case <- taskQuit ://rbmq断开连接后 开始尝试重新建立链接
            fmt.Println("rbmq断开连接后 开始尝试重新建立链接")
             go tryToLink()
        case <- scheduleTimer.C:
            //fmt.Println("~~~~~~~~~~~~~~~~~~~~~~~")
        }
        // 重置调度间隔
        scheduleTimer.Reset(time.Millisecond*300)
        if !exitTask{
            break
        }
    }
    fmt.Println("exit")
    return
}


func Recv2(mq RabbitMQ,receiver Receiver,taskQuit chan<- struct{}){
        defer func() {
            fmt.Println("rbmq链接失败,协程任务退出~~~~~~~~~~~~~~~~~~~~")
            taskQuit <- struct{}{}
            return
        }()
        // 验证链接是否正常
        err := mq.MqOpenChannel()
        if(err != nil){
            return
        }
        mq.ListenReceiver(receiver)
}


type retryPro struct {
    msgContent   string
}
View Code

 

 

核心代码分析： 
/*
runNums  开启并发执行任务数量
 */
func Recv(queueExchange QueueExchange,receiver Receiver,otherParams ...int) (err error){
    var (
        exitTask bool
        maxTryConnNums int  //rbmq链接失败后多久尝试一次
        runNums int
        maxTryConnTimeFromMinute int
    )

    if(len(otherParams) <= 0){
        runNums = 1
        maxTryConnTimeFromMinute = 0
    }else if(len(otherParams) == 1){
        runNums = otherParams[0]
        maxTryConnTimeFromMinute = 0
    }else if(len(otherParams) == 2){
        runNums = otherParams[0]
        maxTryConnTimeFromMinute = otherParams[1]
    }


    //maxTryConnNums := 360 //rbmq链接失败后最大尝试次数
    //maxTryConnTime := time.Duration(10) //rbmq链接失败后多久尝试一次
    maxTryConnNums = maxTryConnTimeFromMinute * 10 * maxTryConnTimeFromMinute//rbmq链接失败后最大尝试次数
    maxTryConnTime := time.Duration(6) //rbmq链接失败后多久尝试一次
    mq := NewMq(queueExchange)
    //链接rabbitMQ
    err = mq.MqConnect()
    if(err != nil){
        return
    }

    defer func() {
        if panicErr := recover(); panicErr != nil{
            fmt.Println(recover())
            err = errors.New(fmt.Sprintf("%s",panicErr))
        }
    }()

    //rbmq断开链接后 协程退出释放信号
    taskQuit:= make(chan struct{}, 1)
    //尝试链接rbmq
    tryToLinkC := make(chan struct{}, 1)

    //最大尝试次数
    tryToLinkMaxNums := make(chan struct{}, 1)

    maxTryNums := 0 //尝试重启次数

    //开始执行任务
    for i:=1;i<=runNums;i++{
        go Recv2(mq,receiver,taskQuit);
    }

    //如果rbmq断开连接后 尝试重新建立链接
    var tryToLink = func() {
        for {
            maxTryNums += 1
            err = mq.MqConnect()
            if(err == nil){
                tryToLinkC <- struct{}{}
                break
            }
            if(maxTryNums > maxTryConnNums){
                tryToLinkMaxNums <- struct{}{}
                break
            }
            //如果链接断开了 10秒重新尝试链接一次
            time.Sleep(time.Second * maxTryConnTime)
        }
        return
    }
    scheduleTimer := time.NewTimer(time.Millisecond*300)
    exitTask = true
    for{
        select {
        case <-tryToLinkC: //建立链接成功后 重新开启协程执行任务
            fmt.Println("重新开启新的协程执行任务")
            go Recv2(mq,receiver,taskQuit);
        case <-tryToLinkMaxNums://rbmq超出最大链接次数 退出任务
            fmt.Println("rbmq链接超过最大尝试次数!")
            exitTask = false
            err = errors.New("rbmq链接超过最大尝试次数!")
        case <- taskQuit ://rbmq断开连接后 开始尝试重新建立链接
            fmt.Println("rbmq断开连接后 开始尝试重新建立链接")
             go tryToLink()
        case <- scheduleTimer.C:
            //fmt.Println("~~~~~~~~~~~~~~~~~~~~~~~")
        }
        // 重置调度间隔
        scheduleTimer.Reset(time.Millisecond*300)
        if !exitTask{
            break
        }
    }
    fmt.Println("exit")
    return
}


func Recv2(mq RabbitMQ,receiver Receiver,taskQuit chan<- struct{}){
        defer func() {
            fmt.Println("rbmq链接失败,协程任务退出~~~~~~~~~~~~~~~~~~~~")
            taskQuit <- struct{}{}
            return
        }()
        // 验证链接是否正常
        err := mq.MqOpenChannel()
        if(err != nil){
            return
        }
        mq.ListenReceiver(receiver)
}

 

 

 

场景一：rbmq断开连接  ，开始尝试链接，链接次数超过最大次数  放弃链接  

 

场景二：rbmq断开连接  ，开始尝试链接，链接次数未超过最大次数，重新建议链接 开始消费任务

 

最新源码仓库地址：https://github.com/sunlongv520/golang-rabbitmq

 

其它：该rabbitmq包实现中包含了，消息处理失败重试机制，有兴趣的同学可以看看

　　（重试和重连接是两个概念）

　　重连接 ：rabbitmq链接失败导致任务失败，此时要等待rabbitmq服务器恢复正常后才能再次启动协程处理任务

　　重试：rabbitmq服务正常，消息消费进程也正常，但是消息处理失败。尝试多次消费消息后还是失败就ack消息，在整个重试过程中不会阻塞消费

 

消息重试机制：https://www.cnblogs.com/sunlong88/p/11982741.html

上一篇：[入门到吐槽系列] 微信小程序 敏感违规图片检测 mediaCheckAsync，客服接口 消息推
下一篇：没有了
网友评论
相关栏目
最近更新
热门文章