From d587d14f86c3ef69ba2c4dd4f82940303ac7d6f0 Mon Sep 17 00:00:00 2001 From: Gordon <1432970085@qq.com> Date: Wed, 8 Jun 2022 17:11:17 +0800 Subject: [PATCH] log --- internal/msg_transfer/logic/online_history_msg_handler.go | 4 +++- internal/msg_transfer/logic/persistent_msg_handler.go | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/internal/msg_transfer/logic/online_history_msg_handler.go b/internal/msg_transfer/logic/online_history_msg_handler.go index 7baccf176..2b6a1bf18 100644 --- a/internal/msg_transfer/logic/online_history_msg_handler.go +++ b/internal/msg_transfer/logic/online_history_msg_handler.go @@ -466,7 +466,9 @@ func (och *OnlineHistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupS //och.chArrays[channelID] <- Cmd2Value{Cmd: UserMessages, Value: MsgChannelValue{userID: userID, msgList: []*pbMsg.MsgDataToMQ{&msgFromMQ}, triggerID: msgFromMQ.OperationID}} //sess.MarkMessage(msg, "") rwLock.Lock() - cMsg = append(cMsg, msg) + if len(msg.Value) != 0 { + cMsg = append(cMsg, msg) + } rwLock.Unlock() sess.MarkMessage(msg, "") //och.TriggerCmd(OnlineTopicBusy) diff --git a/internal/msg_transfer/logic/persistent_msg_handler.go b/internal/msg_transfer/logic/persistent_msg_handler.go index 83ca7bd12..f0104740a 100644 --- a/internal/msg_transfer/logic/persistent_msg_handler.go +++ b/internal/msg_transfer/logic/persistent_msg_handler.go @@ -72,7 +72,9 @@ func (pc *PersistentConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi claim sarama.ConsumerGroupClaim) error { for msg := range claim.Messages() { log.NewDebug("", "kafka get info to mysql", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key)) - pc.msgHandle[msg.Topic](msg, string(msg.Key), sess) + if len(msg.Value) != 0 { + pc.msgHandle[msg.Topic](msg, string(msg.Key), sess) + } sess.MarkMessage(msg, "") } return nil