Ma/origin
1
0
Fork 0
mirror of https://github.com/duanhf2012/origin.git synced 2026-02-04 06:54:45 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

优化消息队列日志

Browse Source
This commit is contained in:
boyce
2025-05-21 21:34:32 +08:00
parent 01d3b3e535
commit ecfd42bdec
5 changed files with 18 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
sysservice/messagequeueservice/CustomerSubscriber.go
View File

@@ -104,11 +104,11 @@ func (cs *CustomerSubscriber) UnSubscribe() {
func (cs *CustomerSubscriber) LoadLastIndex() { func (cs *CustomerSubscriber) LoadLastIndex() {
for { for {
if atomic.LoadInt32(&cs.isStop) != 0 { if atomic.LoadInt32(&cs.isStop) != 0 {
log.Info("topic ", cs.topic, " out of subscription") log.SInfo("topic ", cs.topic, " out of subscription")
break break
} }
log.Info("customer ", cs.customerId, " start load last index ") log.SInfo("customer ", cs.customerId, " start load last index ")
lastIndex, ret := cs.subscriber.dataPersist.LoadCustomerIndex(cs.topic, cs.customerId) lastIndex, ret := cs.subscriber.dataPersist.LoadCustomerIndex(cs.topic, cs.customerId)
if ret == true { if ret == true {
if lastIndex > 0 { if lastIndex > 0 {
@@ -116,18 +116,18 @@ func (cs *CustomerSubscriber) LoadLastIndex() {
} else { } else {
//否则直接使用客户端发回来的 //否则直接使用客户端发回来的
} }
log.Info("customer ", cs.customerId, " load finish,start index is ", cs.StartIndex) log.SInfo("customer ", cs.customerId, " load finish,start index is ", cs.StartIndex)
break break
} }
log.Info("customer ", cs.customerId, " load last index is fail...") log.SInfo("customer ", cs.customerId, " load last index is fail...")
time.Sleep(5 * time.Second) time.Sleep(5 * time.Second)
} }
} }
func (cs *CustomerSubscriber) SubscribeRun() { func (cs *CustomerSubscriber) SubscribeRun() {
defer cs.subscriber.queueWait.Done() defer cs.subscriber.queueWait.Done()
log.Info("topic ", cs.topic, " start subscription") log.SInfo("topic ", cs.topic, " start subscription")
//加载之前的位置 //加载之前的位置
if cs.subscribeMethod == MethodLast { if cs.subscribeMethod == MethodLast {
@@ -136,7 +136,7 @@ func (cs *CustomerSubscriber) SubscribeRun() {
for { for {
if atomic.LoadInt32(&cs.isStop) != 0 { if atomic.LoadInt32(&cs.isStop) != 0 {
log.Info("topic ", cs.topic, " out of subscription") log.SInfo("topic ", cs.topic, " out of subscription")
break break
} }
@@ -146,14 +146,14 @@ func (cs *CustomerSubscriber) SubscribeRun() {
//todo 检测退出 //todo 检测退出
if cs.subscribe() == false { if cs.subscribe() == false {
log.Info("topic ", cs.topic, " out of subscription") log.SInfo("topic ", cs.topic, " out of subscription")
break break
} }
} }
//删除订阅关系 //删除订阅关系
cs.subscriber.removeCustomer(cs.customerId, cs) cs.subscriber.removeCustomer(cs.customerId, cs)
log.Info("topic ", cs.topic, " unsubscription") log.SInfo("topic ", cs.topic, " unsubscription")
} }
func (cs *CustomerSubscriber) subscribe() bool { func (cs *CustomerSubscriber) subscribe() bool {

4
sysservice/messagequeueservice/MessageQueueService.go
View File

@@ -63,7 +63,7 @@ func (ms *MessageQueueService) ReadCfg() error {
maxProcessTopicBacklogNum, ok := mapDBServiceCfg["MaxProcessTopicBacklogNum"] maxProcessTopicBacklogNum, ok := mapDBServiceCfg["MaxProcessTopicBacklogNum"]
if ok == false { if ok == false {
ms.maxProcessTopicBacklogNum = DefaultMaxTopicBacklogNum ms.maxProcessTopicBacklogNum = DefaultMaxTopicBacklogNum
log.Info("MaxProcessTopicBacklogNum config is set to the default value of ", maxProcessTopicBacklogNum) log.SInfo("MaxProcessTopicBacklogNum config is set to the default value of ", maxProcessTopicBacklogNum)
} else { } else {
ms.maxProcessTopicBacklogNum = int32(maxProcessTopicBacklogNum.(float64)) ms.maxProcessTopicBacklogNum = int32(maxProcessTopicBacklogNum.(float64))
} }
@@ -71,7 +71,7 @@ func (ms *MessageQueueService) ReadCfg() error {
memoryQueueLen, ok := mapDBServiceCfg["MemoryQueueLen"] memoryQueueLen, ok := mapDBServiceCfg["MemoryQueueLen"]
if ok == false { if ok == false {
ms.memoryQueueLen = DefaultMemoryQueueLen ms.memoryQueueLen = DefaultMemoryQueueLen
log.Info("MemoryQueueLen config is set to the default value of ", DefaultMemoryQueueLen) log.SInfo("MemoryQueueLen config is set to the default value of ", DefaultMemoryQueueLen)
} else { } else {
ms.memoryQueueLen = int32(memoryQueueLen.(float64)) ms.memoryQueueLen = int32(memoryQueueLen.(float64))
} }

6
sysservice/messagequeueservice/MongoPersist.go
View File

@@ -237,7 +237,7 @@ func (mp *MongoPersist) findTopicData(topic string, startIndex uint64, limit int
defer cancelAll() defer cancelAll()
err = cursor.All(ctxAll, &res) err = cursor.All(ctxAll, &res)
if err != nil { if err != nil {
log.Error("find collect name ", topic, " is error", log.ErrorAttr("err", err)) log.Error("find collect name error",log.String("topic",topic) ,log.ErrorField("err",err))
return nil, false return nil, false
} }
@@ -246,7 +246,7 @@ func (mp *MongoPersist) findTopicData(topic string, startIndex uint64, limit int
rawData, errM := bson.Marshal(res[i]) rawData, errM := bson.Marshal(res[i])
if errM != nil { if errM != nil {
if errM != nil { if errM != nil {
log.Error("collect name ", topic, " Marshal is error", log.ErrorAttr("err", err)) log.Error("Marshal error",log.String("topic",topic) , log.ErrorField("err", err))
return nil, false return nil, false
} }
continue continue
@@ -391,7 +391,7 @@ func (mp *MongoPersist) GetIndex(topicData *TopicData) uint64 {
if e.Key == "_id" { if e.Key == "_id" {
errC, seq := convertToNumber[uint64](e.Value) errC, seq := convertToNumber[uint64](e.Value)
if errC != nil { if errC != nil {
log.Error("value is error:%s,%+v, ", errC.Error(), e.Value) log.Error("value is error", log.ErrorField("err",errC), log.Any("val",e.Value))
} }
return seq return seq

6
sysservice/messagequeueservice/Subscriber.go
View File

@@ -56,9 +56,9 @@ func (ss *Subscriber) TopicSubscribe(rpcHandler rpc.IRpcHandler, subScribeType r
} }
if ok == true { if ok == true {
log.Info("repeat subscription for customer ", customerId) log.SInfo("repeat subscription for customer ", customerId)
} else { } else {
log.Info("subscription for customer ", customerId) log.SInfo("subscription for customer ", customerId)
} }
} }
@@ -72,7 +72,7 @@ func (ss *Subscriber) UnSubscribe(customerId string) {
customerSubscriber, ok := ss.mapCustomer[customerId] customerSubscriber, ok := ss.mapCustomer[customerId]
if ok == false { if ok == false {
log.SWarning("failed to unsubscribe customer " + customerId) log.SWarn("failed to unsubscribe customer ", customerId)
return return
} }

4
sysservice/messagequeueservice/TopicRoom.go
View File

@@ -93,7 +93,7 @@ func (tr *TopicRoom) Stop() {
func (tr *TopicRoom) topicRoomRun() { func (tr *TopicRoom) topicRoomRun() {
defer tr.queueWait.Done() defer tr.queueWait.Done()
log.Info("topic room ", tr.topic, " is running..") log.SInfo("topic room ", tr.topic, " is running..")
for { for {
if atomic.LoadInt32(&tr.isStop) != 0 { if atomic.LoadInt32(&tr.isStop) != 0 {
break break
@@ -145,5 +145,5 @@ func (tr *TopicRoom) topicRoomRun() {
} }
tr.customerLocker.Unlock() tr.customerLocker.Unlock()
log.Info("topic room ", tr.topic, " is stop") log.SInfo("topic room ", tr.topic, " is stop")
} }