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

concurrent consumption of messages

Browse Source
This commit is contained in:
Gordon 2022-05-20 21:58:23 +08:00
parent 55907a9a05
commit 01b065b3c2
1 changed files with 18 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -392,26 +392,11 @@ func (och *OnlineHistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupS
cMsg := make([]*sarama.ConsumerMessage, 0, 500) cMsg := make([]*sarama.ConsumerMessage, 0, 500)
t := time.NewTicker(time.Duration(100) * time.Millisecond) t := time.NewTicker(time.Duration(100) * time.Millisecond)
var triggerID string var triggerID string
for msg := range claim.Messages() { for {
//och.TriggerCmd(OnlineTopicBusy) //och.TriggerCmd(OnlineTopicBusy)
cMsg = append(cMsg, msg)
select { select {
case <-t.C: case msg := <-claim.Messages():
if len(cMsg) >= 0 { cMsg = append(cMsg, msg)
ccMsg := make([]*sarama.ConsumerMessage, 0, 500)
for _, v := range cMsg {
ccMsg = append(ccMsg, v)
}
triggerID = utils.OperationIDGenerator()
log.Debug(triggerID, "timer trigger msg consumer start", len(ccMsg))
och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
triggerID: triggerID, cmsgList: ccMsg}}
sess.MarkMessage(msg, "")
cMsg = make([]*sarama.ConsumerMessage, 0, 500)
log.Debug(triggerID, "timer trigger msg consumer end", len(cMsg))
}
default:
if len(cMsg) >= 500 { if len(cMsg) >= 500 {
ccMsg := make([]*sarama.ConsumerMessage, 0, 500) ccMsg := make([]*sarama.ConsumerMessage, 0, 500)
for _, v := range cMsg { for _, v := range cMsg {
@ -426,9 +411,23 @@ func (och *OnlineHistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupS
cMsg = make([]*sarama.ConsumerMessage, 0, 500) cMsg = make([]*sarama.ConsumerMessage, 0, 500)
log.Debug(triggerID, "length trigger msg consumer end", len(cMsg)) log.Debug(triggerID, "length trigger msg consumer end", len(cMsg))
} }
case <-t.C:
if len(cMsg) > 0 {
ccMsg := make([]*sarama.ConsumerMessage, 0, 500)
for _, v := range cMsg {
ccMsg = append(ccMsg, v)
}
triggerID = utils.OperationIDGenerator()
log.Debug(triggerID, "timer trigger msg consumer start", len(ccMsg))
och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
triggerID: triggerID, cmsgList: ccMsg}}
sess.MarkMessage(cMsg[len(cMsg)-1], "")
cMsg = make([]*sarama.ConsumerMessage, 0, 500)
log.Debug(triggerID, "timer trigger msg consumer end", len(cMsg))
}
} }
log.NewDebug("", "online kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "online", msg.Offset, claim.HighWaterMarkOffset()) //log.NewDebug("", "online kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "online", msg.Offset, claim.HighWaterMarkOffset())
} }
return nil return nil