golang实现rabbitmq消费者模式 断线重连机制

在连接第三方组件的时候实现断线重连机制是很有必要的事情,因为你不知道在啥时候他突然就抽风挂掉了。以rabbitmq为例,在突然面对大流量写入,或者连接数被打满(比如在资讯,直播等模块使用rabbitmq(惨痛的教训))mq就挂掉了,如果你的消费者没有重连机制,你的消费者进程在mq挂掉后,也会自动挂掉。然后等运维修复了mq,但是你的消费者进程却无法再次连接消费了,这就有点恐怖了。要不就写个shell 脚本 ,每隔30S或者1分钟去检测进程活度,挂掉了就重启消费者。当然最简单的还是在消费者的代码里实现断线重连机制。代码如下

package rabbitmq

import (
   "encoding/json"
   "fmt"
   "github.com/streadway/amqp"
   "time"
)

//测试用例
type Obj struct {
   Item1 string `json:"item1"`
   Item2 string `json:"item2"`
   Item3 string `json:"item3"`
}

func StartAMQPConsume() {
   defer func() {
      if err := recover(); err != nil {
         time.Sleep(3 * time.Second)
         fmt.Println("休息3秒")
         StartAMQPConsume()
      }
   }()
   conn, err := amqp.Dial("amqp://guest:guest@127.0.0.1:5672/") //mq连接地址
   if err != nil {
      fmt.Println(err)
   }
   defer conn.Close()
   ch, err := conn.Channel()
   if err != nil {
      fmt.Println(err)
   }
   defer ch.Close()
   closeChan := make(chan *amqp.Error, 1)
   notifyClose := ch.NotifyClose(closeChan) //一旦消费者的channel有错误,产生一个amqp.Error,channel监听并捕捉到这个错误
   closeFlag := false
   msgs, err := ch.Consume(
      "xly.test.queue",
      "",
      true,
      false,
      false,
      false, nil)
   var obj Obj
   for {
      select {
      case e := <-notifyClose:
         fmt.Println("chan通道错误,e:%s", e.Error())
         close(closeChan)
         time.Sleep(5 * time.Second)
         StartAMQPConsume()
         closeFlag = true
      case msg := <-msgs:
         //fmt.Println()
         if err := json.Unmarshal(msg.Body, &obj); err != nil {

            fmt.Println(err.Error())
         }
         fmt.Println(obj.Item1)

      }
      if closeFlag {
         break
      }
   }
}

只需要在main 方法里 调用这个方法即可,或者go协程调用 ,效果如下

golang实现rabbitmq消费者模式 断线重连机制当我关闭mq服务时会有如下效果

golang实现rabbitmq消费者模式 断线重连机制

再次启动mq

golang实现rabbitmq消费者模式 断线重连机制

以上就是rabbitmq 消费者断线重连的完整代码

上一篇:Spring AMQP:拦截所有传入和传出消息


下一篇:Spring AMQP:RabbitTemplate SimpleMessageListenerContainer