rabbitmq golang consumer 断开重连

(1) rabbitmq 官网建议,多线程尽量共享连接，独享信道。共享信道不是线程安全的。先来看下rabbitmq的粗糙结构图。

image.png

(2)消费者部分官方代码示例

msgs, err := ch.Consume(
               q.Name, // queue
               "",     // consumer
               true,   // auto ack
               false,  // exclusive
               false,  // no local
               false,  // no wait
               nil,    // args
       )
       failOnError(err, "Failed to register a consumer")

       forever := make(chan bool)

       go func() {
               for d := range msgs {
                       log.Printf(" [x] %s", d.Body)
               }
       }()

       log.Printf(" [*] Waiting for logs. To exit press CTRL+C")
       <-forever

真正消费的地方在 go func()中，当msgs有数据传递过来，就消费，没有就等待，那么问题来了，
怎么检测是否连接正常的呢？当连接断开或者 channel 关闭，msgs是不会被close的还好官方提供了检测机制，请移步下一关卡：
(3)NotifyClose函数

msgs, err := channel.Consume(
       queue_name, // queue
       "",        // consumer
       false,      // auto-ack
       false,      // exclusive
       false,      // no-local
       false,      // no-wait
       nil,        // args
)
if err != nil {
       return err
}
closeChan := make(chan bool, 1)
go func(chan<- bool) {
      cc := make(chan *amqp.Error)
      e := <-channel.NotifyClose(cc)
      seelog.Trace("[RABBITMQ_CLIENT]", "channel close error:", e.Error())
      closeChan <- true
}(closeChan)

go func(<-chan bool){
    for{
        select{
                case d := <-msgs:
                       log.Printf(" [x] %s", d.Body)
                case <-closeChan:
                        return
        }
    }
}(closeChan)

为什么这样就可以收到通知了呢？我们来看下源码

type Connection struct {
...
noNotify bool 
closes []chan *Error 
... 
}

func (c *Connection) NotifyClose(receiver chan *Error) chan *Error {
  c.m.Lock()
  defer c.m.Unlock()

  if c.noNotify {
        close(receiver)
  } else {
        c.closes = append(c.closes, receiver)
  }

  return receiver
}

type Channel struct {
...
noNotify bool 
closes []chan *Error 
... 
}
func (ch *Channel) NotifyClose(c chan *Error) chan *Error {
  ch.notifyM.Lock()
  defer ch.notifyM.Unlock()

  if ch.noNotify {
        close(c)
  } else {
        ch.closes = append(ch.closes, c)
  }

  return c
}

channel 会向当前channel struct 注册 connection 也会向connection 注册，

func (c *Connection) shutdown(err *Error) {
  atomic.StoreInt32(&c.closed, 1)

        for _, c := range c.closes {
                c <- err
        }

  for _, ch := range c.channels {
        ch.shutdown(err)
  }
}

func (ch *Channel) shutdown(e *Error) {

  for _, c := range ch.closes {
        c <- e
  }
  atomic.StoreInt32(&ch.closed, 1)
}

当程序检测到当前连接或者当前通道异常的时候，会调用相应的shutdown方法，那么所有向 closes []chan *Error 注册的chan 都会收到消息。
当连接断开的时候，所有的通道都会被通知到，当通道异常关闭的时候，只需要通知当前通道即可。
上面我们使用的方法是，注册通道错误通道，异常通知
（4）重连接
上述，我们不知道是否只是信道异常还是，连接异常，我们可以用IsClosed函数来判断，shutdown 连接的时候，会把c.closed 标置为1，IsClosed函数就是用来检测这个的，如果是连接已经关闭了就重连，如果是信道异常，就再获取一个新的信道

func (c *Connection) IsClosed() bool {
  return (atomic.LoadInt32(&c.closed) == 1)
}

本人写的rabbitmq 连接池尚不成熟，不足以见世面。

最后编辑于：2019.04.10 13:04:31