blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-25 11:06:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

concurrent consumption of messages

Browse Source
This commit is contained in:
Gordon 2022-05-20 17:32:40 +08:00
parent a12c8f4dbc
commit c5f055718a
1 changed files with 4 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
internal/msg_transfer/logic/online_history_msg_handler.go
View File

@ -209,8 +209,8 @@ func (och *OnlineHistoryConsumerHandler) Run(channelID int) {
//}
func (och *OnlineHistoryConsumerHandler) MessagesDistributionHandle() {
UserAggregationMsgs := make(map[string][]*pbMsg.MsgDataToMQ, ChannelNum)
for {
UserAggregationMsgs := make(map[string][]*pbMsg.MsgDataToMQ, ChannelNum)
select {
case cmd := <-och.msgDistributionCh:
switch cmd.Cmd {
@ -250,6 +250,7 @@ func (och *OnlineHistoryConsumerHandler) MessagesDistributionHandle() {
}
}
}
}
}
@ -403,7 +404,7 @@ func (och *OnlineHistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupS
}
triggerID = utils.OperationIDGenerator()
log.Debug(triggerID, "timer trigger msg consumer start", len(cMsg))
log.Debug(triggerID, "timer trigger msg consumer start", len(ccMsg))
och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
triggerID: triggerID, cmsgList: ccMsg}}
sess.MarkMessage(msg, "")
@ -418,7 +419,7 @@ func (och *OnlineHistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupS
}
triggerID = utils.OperationIDGenerator()
log.Debug(triggerID, "length trigger msg consumer start", len(cMsg))
log.Debug(triggerID, "length trigger msg consumer start", len(ccMsg))
och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
triggerID: triggerID, cmsgList: ccMsg}}
sess.MarkMessage(msg, "")