This commit is contained in:
laodaming 2023-07-28 15:19:48 +08:00
parent 3e703cf85a
commit 9c3d9c06b3
3 changed files with 11 additions and 26 deletions

View File

@ -18,6 +18,7 @@ type ServiceContext struct {
{{.middleware}} {{.middleware}}
MysqlConn *gorm.DB MysqlConn *gorm.DB
AllModels *gmodel.AllModelsGen AllModels *gmodel.AllModelsGen
RabbitMq map[string]*initalize.QueueItem
} }
func NewServiceContext(c {{.config}}) *ServiceContext { func NewServiceContext(c {{.config}}) *ServiceContext {
@ -26,6 +27,7 @@ func NewServiceContext(c {{.config}}) *ServiceContext {
Config: c, Config: c,
MysqlConn: initalize.InitMysql(c.SourceMysql), MysqlConn: initalize.InitMysql(c.SourceMysql),
AllModels: gmodel.NewAllModels(initalize.InitMysql(c.SourceMysql)), AllModels: gmodel.NewAllModels(initalize.InitMysql(c.SourceMysql)),
RabbitMq: initalize.InitRabbitMq(c.SourceRabbitMq, nil),
{{.middlewareAssignment}} {{.middlewareAssignment}}
} }
} }

View File

@ -5,27 +5,25 @@ import (
"fusenapi/constants" "fusenapi/constants"
"github.com/streadway/amqp" "github.com/streadway/amqp"
"log" "log"
"time"
) )
// handle // handle
type queueItem struct { type QueueItem struct {
Ch *amqp.Channel Ch *amqp.Channel
Queue amqp.Queue Queue amqp.Queue
} }
var mapMq = make(map[string]*queueItem) func InitRabbitMq(url string, config *tls.Config) map[string]*QueueItem {
func InitRabbitMq(url string, config *tls.Config) (map[string]*queueItem, error) {
conn, err := amqp.DialTLS(url, config) conn, err := amqp.DialTLS(url, config)
if err != nil { if err != nil {
return nil, err log.Fatalf("Failed to connect to RabbitMQ: %v", err)
} }
// 创建一个通道 // 创建一个通道
ch, err := conn.Channel() ch, err := conn.Channel()
if err != nil { if err != nil {
return nil, err log.Fatalf("Failed to open a channel: %v", err)
} }
mapMq := make(map[string]*QueueItem)
//声明队列 //声明队列
for _, queueName := range constants.MQ_QUEUE_ARR { for _, queueName := range constants.MQ_QUEUE_ARR {
q, err := ch.QueueDeclare( q, err := ch.QueueDeclare(
@ -41,27 +39,10 @@ func InitRabbitMq(url string, config *tls.Config) (map[string]*queueItem, error)
ch.Close() ch.Close()
log.Fatalf("Failed to declare a queue: %v", err) log.Fatalf("Failed to declare a queue: %v", err)
} }
mapMq[string(queueName)] = &queueItem{ mapMq[string(queueName)] = &QueueItem{
Ch: ch, Ch: ch,
Queue: q, Queue: q,
} }
} }
go checkAlive(url, config, conn, ch) return mapMq
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
}
}
}
} }

View File

@ -18,6 +18,7 @@ type ServiceContext struct {
MysqlConn *gorm.DB MysqlConn *gorm.DB
AllModels *gmodel.AllModelsGen AllModels *gmodel.AllModelsGen
RabbitMq map[string]*initalize.QueueItem
} }
func NewServiceContext(c config.Config) *ServiceContext { func NewServiceContext(c config.Config) *ServiceContext {
@ -26,6 +27,7 @@ func NewServiceContext(c config.Config) *ServiceContext {
Config: c, Config: c,
MysqlConn: initalize.InitMysql(c.SourceMysql), MysqlConn: initalize.InitMysql(c.SourceMysql),
AllModels: gmodel.NewAllModels(initalize.InitMysql(c.SourceMysql)), AllModels: gmodel.NewAllModels(initalize.InitMysql(c.SourceMysql)),
RabbitMq: initalize.InitRabbitMq(c.SourceRabbitMq, nil),
} }
} }