AMQP重连机制实现

// producer.go

// NewClient .
func NewClient(cfg *types.RabbitMQConfig) *Client {
    // init MQ connection
	return &Client{
		ch:  ch, // *amqp.Channel
		cfg: cfg,
	}
}

// Client .
type Client struct {
	ch  *amqp.Channel
	cfg *types.RabbitMQConfig
}

// Send . send by routing 
func (c *Client) Send(payload *types.Payload) error {
	var routing string

	switch payload.Typ {
	case types.PayloadTypUsers:
		routing = c.cfg.MqRoutingUser
	default:
		routing = c.cfg.MqRoutingSys
	}

	dat, _ := json.Marshal(payload)
	publishing := amqp.Publishing{
		ContentType: "text/plain",
		Body:        dat,
	}

	return c.ch.Publish(c.cfg.MqExchagne, routing, false, false, publishing)
}

// consumer.go
// ...
func (c *Client) Consume() error {
    err := c.initRabbitmqResource()
	if err != nil {
		logger.Std.Errorf("could init rabbitmq resource: %v", err)
		return err
	}

	msgs, err := c.rabbitmqChannel.Consume(
		c.rabbitmqQ, // queue
		"",          // consumer
		false,       // auto ack
		false,       // exclusive
		false,       // no local
		false,       // no wait
		nil,         // args
	)
	if err != nil {
		logger.Std.Errorf("could not consume messages: %v", err)
		return err
	}

	logger.Std.Infof("start consumming messages")
	for d := range msgs {
		logger.Std.Debugf("consumming msg: body=%s, ex=%s, routing=%s, content-type=%s",
			d.Body, d.Exchange, d.RoutingKey, d.ContentType)
		if err := c.sendout(d.Body); err != nil {
			continue
		}

		if err := d.Ack(false); err != nil {
			logger.Std.Errorf("ack message error: %v", err)
		}
	}

	// 队列被删除时会被执行到
	logger.Std.Infof("this should not be executed !!!!!")
	return nil
}
// ...

func (subs *consumers) buffer(in chan *Delivery, out chan Delivery) {
	// 退出时关闭消费channel
	defer close(out)
	defer subs.Done()

	var inflight = in
	var queue []*Delivery

	for delivery := range in {
		queue = append(queue, delivery)

		for len(queue) > 0 {
			select {
			// 很明显，协程会在消费者被关闭的时候退出
			case <-subs.closed:
				// closed before drained, drop in-flight
				return

			case delivery, consuming := <-inflight:
				if consuming {
					queue = append(queue, delivery)
				} else {
					inflight = nil
				}

			case out <- *queue[0]:
				queue = queue[1:]
			}
		}
	}
}

func (w *Wrapper) handleReconnect() {
	for {
		// 未连接上的情况下一直尝试连接
		if !w.isConnected {
			log.Println("Attempting to connect")
			var (
				connected = false
				err       error
			)

			for cnt := 0; !connected; cnt++ {
				// connect 尝试连接，并在成功后触发changeConn事件，也会更新isConnected
				if connected, err = w.connect(); err != nil {
					log.Printf("Failed to connect: %s.\n", err)
				}
				if !connected {
					log.Printf("Retrying... %d\n", cnt)
				}
				time.Sleep(reconnectDelay)
			}
		}

		// 连接正常的情况下，处理channel.NotifyClose 和 conn.NotifyClose
		// 注：在任意一个notify事件通知后，应完全处理两个channel中的消息，否则会造成无缓冲阻塞。
		// https://ms2008.github.io/2019/06/16/golang-rabbitmq/
		select {
		case <-w.done:
			println("w.done")
			return
		case err := <-w.chNotify:
			log.Printf("channel close notify: %v", err)
			w.isConnected = false
		case err := <-w.connNotify:
			log.Printf("conn close notify: %v", err)
			w.isConnected = false
		}
		time.Sleep(reconnectDetectDur)
	}
}