xinyin025/hotgo
1
0
Fork 0
mirror of https://github.com/bufanyun/hotgo.git synced 2025-10-12 04:53:47 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

更新2.1.2版本，优化部门、角色权限，增加上下级关系；增加登录、系统、短信日志;优化省市区编码

Browse Source
This commit is contained in:
孟帅
2023-01-25 11:49:21 +08:00
parent 11fad0132d
commit 93e0fe7250
190 changed files with 35896 additions and 7208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
server/internal/library/queue/kafkamq.go
View File

@@ -87,30 +87,28 @@ func (r *KafkaMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg))
}
ctx, cancel := context.WithCancel(context.Background())
go func() {
go func(ctx context.Context) {
for {
if err := r.consumerIns.Consume(ctx, []string{topic}, &consumer); err != nil {
FatalLog(ctx, "kafka Error from consumer", err)
g.Log().Fatalf(ctx, "kafka Error from consumer, err%+v", err)
}
if ctx.Err() != nil {
Log(ctx, fmt.Sprintf("kafka consoumer stop : %v", ctx.Err()))
g.Log().Debugf(ctx, fmt.Sprintf("kafka consoumer stop : %v", ctx.Err()))
return
}
consumer.ready = make(chan bool)
}
}()
}(ctx)
<-consumer.ready // Await till the consumer has been set up
Log(ctx, "kafka consumer up and running!...")
g.Log().Debug(ctx, "kafka consumer up and running!...")
signal.AppDefer(func() {
Log(ctx, "kafka consumer close...")
g.Log().Debug(ctx, "kafka consumer close...")
cancel()
if err = r.consumerIns.Close(); err != nil {
FatalLog(ctx, "kafka Error closing client", err)
g.Log().Fatalf(ctx, "kafka Error closing client, err:%+v", err)
}
})
@@ -193,7 +191,7 @@ func RegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) {
}
signal.AppDefer(func() {
Log(ctx, "kafka producer AsyncClose...")
g.Log().Debug(ctx, "kafka producer AsyncClose...")
mqIns.producerIns.AsyncClose()
})
}