更新重连

rabbitmq.go

@@ -28,9 +28,13 @@ type Config struct {
 }
 
 // InitRabbitmq @Title 初始化rabbitmq
-func InitRabbitmq(config Config) (err error) {
+func InitRabbitmq(config Config) error {
 	Rabbitmq.config = config
-	return getConn()
+	err := getConn()
+	if err != nil && Rabbitmq.config.ReConn {
+		go reConn()
+	}
+	return err
 }
 
 // @Title 获取连接
@@ -108,6 +112,8 @@ func (r *rabbitmq) Publish(exchangeName, msg string, delay ...int64) error {
 		amqp.Publishing{
 			ContentType:  "text/plain",
 			Body:         []byte(msg),
+			DeliveryMode: amqp.Persistent,
+			Timestamp:    time.Now(),
 			Headers:      header,
 		})
 }
@@ -134,6 +140,8 @@ func (r *rabbitmq) PublishWithContext(ctx context.Context, exchangeName, key, ms
 		false,        //立即
 		amqp.Publishing{
 			ContentType:  "text/plain",
+			DeliveryMode: amqp.Persistent,
+			Timestamp:    time.Now(),
 			Body:         []byte(msg),
 			Headers:      header,
 		})