fix

initalize/rabbitmq.go

@@ -0,0 +1,67 @@
+package initalize
+
+import (
+	"crypto/tls"
+	"fusenapi/constants"
+	"github.com/streadway/amqp"
+	"log"
+	"time"
+)
+
+// handle
+type queueItem struct {
+	Ch    *amqp.Channel
+	Queue amqp.Queue
+}
+
+var mapMq = make(map[string]*queueItem)
+
+func InitRabbitMq(url string, config *tls.Config) (map[string]*queueItem, error) {
+	conn, err := amqp.DialTLS(url, config)
+	if err != nil {
+		return nil, err
+	}
+	// 创建一个通道
+	ch, err := conn.Channel()
+	if err != nil {
+		return nil, err
+	}
+	//声明队列
+	for _, queueName := range constants.MQ_QUEUE_ARR {
+		q, err := ch.QueueDeclare(
+			string(queueName), // 队列名
+			true,              // 是否持久化
+			false,             // 是否自动删除
+			false,             // 是否排他
+			false,             // 是否等待服务器响应
+			nil,               // 其他参数
+		)
+		if err != nil {
+			conn.Close()
+			ch.Close()
+			log.Fatalf("Failed to declare a queue: %v", err)
+		}
+		mapMq[string(queueName)] = &queueItem{
+			Ch:    ch,
+			Queue: q,
+		}
+	}
+	go checkAlive(url, config, conn, ch)
+	return mapMq, nil
+}
+
+func checkAlive(url string, config *tls.Config, conn *amqp.Connection, ch *amqp.Channel) {
+	var err error
+	for {
+		time.Sleep(time.Second * 1)
+		//断开重连
+		if conn.IsClosed() {
+			mapMq, err = InitRabbitMq(url, config)
+			if err == nil {
+				return
+			} else {
+				continue
+			}
+		}
+	}
+}