optimize seq, specifying that seq is stored in the only certain unknown

internal/msgtransfer/init.go

@@ -0,0 +1,110 @@
+package msgtransfer
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/relation"
+	relationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/tx"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/unrelation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mw"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome"
+	openKeeper "github.com/OpenIMSDK/Open-IM-Server/pkg/discoveryregistry/zookeeper"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/rpcclient"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+type MsgTransfer struct {
+	persistentCH   *PersistentConsumerHandler         // 聊天记录持久化到mysql的消费者 订阅的topic: ws2ms_chat
+	historyCH      *OnlineHistoryRedisConsumerHandler // 这个消费者聚合消息, 订阅的topic：ws2ms_chat, 修改通知发往msg_to_modify topic, 消息存入redis后Incr Redis, 再发消息到ms2pschat topic推送， 发消息到msg_to_mongo topic持久化
+	historyMongoCH *OnlineHistoryMongoConsumerHandler // mongoDB批量插入, 成功后删除redis中消息，以及处理删除通知消息删除的 订阅的topic: msg_to_mongo
+	modifyCH       *ModifyMsgConsumerHandler          // 负责消费修改消息通知的consumer, 订阅的topic: msg_to_modify
+}
+
+func StartTransfer(prometheusPort int) error {
+	db, err := relation.NewGormDB()
+	if err != nil {
+		return err
+	}
+	if err := db.AutoMigrate(&relationTb.ChatLogModel{}); err != nil {
+		return err
+	}
+	rdb, err := cache.NewRedis()
+	if err != nil {
+		return err
+	}
+	mongo, err := unrelation.NewMongo()
+	if err != nil {
+		return err
+	}
+	if err := mongo.CreateMsgIndex(); err != nil {
+		return err
+	}
+	client, err := openKeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema,
+		openKeeper.WithFreq(time.Hour), openKeeper.WithRoundRobin(), openKeeper.WithUserNameAndPassword(config.Config.Zookeeper.Username,
+			config.Config.Zookeeper.Password), openKeeper.WithTimeout(10), openKeeper.WithLogger(log.NewZkLogger()))
+	if err != nil {
+		return err
+	}
+	if client.CreateRpcRootNodes(config.GetServiceNames()); err != nil {
+		return err
+	}
+	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	msgModel := cache.NewMsgCacheModel(rdb)
+	msgDocModel := unrelation.NewMsgMongoDriver(mongo.GetDatabase())
+	extendMsgModel := unrelation.NewExtendMsgSetMongoDriver(mongo.GetDatabase())
+	extendMsgCache := cache.NewExtendMsgSetCacheRedis(rdb, extendMsgModel, cache.GetDefaultOpt())
+	chatLogDatabase := controller.NewChatLogDatabase(relation.NewChatLogGorm(db))
+	extendMsgDatabase := controller.NewExtendMsgDatabase(extendMsgModel, extendMsgCache, tx.NewMongo(mongo.GetClient()))
+	msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel)
+	conversationRpcClient := rpcclient.NewConversationRpcClient(client)
+	groupRpcClient := rpcclient.NewGroupRpcClient(client)
+	msgTransfer := NewMsgTransfer(chatLogDatabase, extendMsgDatabase, msgDatabase, &conversationRpcClient, &groupRpcClient)
+	msgTransfer.initPrometheus()
+	return msgTransfer.Start(prometheusPort)
+}
+
+func NewMsgTransfer(chatLogDatabase controller.ChatLogDatabase,
+	extendMsgDatabase controller.ExtendMsgDatabase, msgDatabase controller.CommonMsgDatabase,
+	conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient) *MsgTransfer {
+	return &MsgTransfer{persistentCH: NewPersistentConsumerHandler(chatLogDatabase), historyCH: NewOnlineHistoryRedisConsumerHandler(msgDatabase, conversationRpcClient, groupRpcClient),
+		historyMongoCH: NewOnlineHistoryMongoConsumerHandler(msgDatabase), modifyCH: NewModifyMsgConsumerHandler(extendMsgDatabase)}
+}
+
+func (m *MsgTransfer) initPrometheus() {
+	prome.NewSeqGetSuccessCounter()
+	prome.NewSeqGetFailedCounter()
+	prome.NewSeqSetSuccessCounter()
+	prome.NewSeqSetFailedCounter()
+	prome.NewMsgInsertRedisSuccessCounter()
+	prome.NewMsgInsertRedisFailedCounter()
+	prome.NewMsgInsertMongoSuccessCounter()
+	prome.NewMsgInsertMongoFailedCounter()
+}
+
+func (m *MsgTransfer) Start(prometheusPort int) error {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	fmt.Println("start msg transfer", "prometheusPort:", prometheusPort)
+	if config.Config.ChatPersistenceMysql {
+		// go m.persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(m.persistentCH)
+	} else {
+		fmt.Println("msg transfer not start mysql consumer")
+	}
+	go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.historyCH)
+	go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.historyMongoCH)
+	go m.modifyCH.modifyMsgConsumerGroup.RegisterHandleAndConsumer(m.modifyCH)
+	err := prome.StartPrometheusSrv(prometheusPort)
+	if err != nil {
+		return err
+	}
+	wg.Wait()
+	return nil
+}

internal/msgtransfer/modify_msg_handler.go

@@ -0,0 +1,113 @@
+package msgtransfer
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller"
+	unRelationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
+	kfk "github.com/OpenIMSDK/Open-IM-Server/pkg/common/kafka"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
+	pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+	"github.com/Shopify/sarama"
+
+	"google.golang.org/protobuf/proto"
+)
+
+type ModifyMsgConsumerHandler struct {
+	modifyMsgConsumerGroup *kfk.MConsumerGroup
+
+	extendMsgDatabase controller.ExtendMsgDatabase
+	extendSetMsgModel unRelationTb.ExtendMsgSetModel
+}
+
+func NewModifyMsgConsumerHandler(database controller.ExtendMsgDatabase) *ModifyMsgConsumerHandler {
+	return &ModifyMsgConsumerHandler{
+		modifyMsgConsumerGroup: kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
+			OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.MsgToModify.Topic},
+			config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToModify),
+		extendMsgDatabase: database,
+	}
+}
+
+func (ModifyMsgConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
+func (ModifyMsgConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
+func (mmc *ModifyMsgConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
+	claim sarama.ConsumerGroupClaim) error {
+	for msg := range claim.Messages() {
+		ctx := mmc.modifyMsgConsumerGroup.GetContextFromMsg(msg)
+		log.ZDebug(ctx, "kafka get info to mysql", "ModifyMsgConsumerHandler", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
+		if len(msg.Value) != 0 {
+			mmc.ModifyMsg(ctx, msg, string(msg.Key), sess)
+		} else {
+			log.ZError(ctx, "msg get from kafka but is nil", nil, "key", msg.Key)
+		}
+		sess.MarkMessage(msg, "")
+	}
+	return nil
+}
+
+func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
+	msgFromMQ := pbMsg.MsgDataToModifyByMQ{}
+	operationID := mcontext.GetOperationID(ctx)
+	err := proto.Unmarshal(cMsg.Value, &msgFromMQ)
+	if err != nil {
+		log.ZError(ctx, "msg_transfer Unmarshal msg err", err, "msg", string(cMsg.Value))
+		return
+	}
+	log.ZDebug(ctx, "proto.Unmarshal MsgDataToMQ", "msgs", msgFromMQ.String())
+	for _, msg := range msgFromMQ.Messages {
+		isReactionFromCache := utils.GetSwitchFromOptions(msg.Options, constant.IsReactionFromCache)
+		if !isReactionFromCache {
+			continue
+		}
+		ctx = mcontext.SetOperationID(ctx, operationID)
+		if msg.ContentType == constant.ReactionMessageModifier {
+			notification := &sdkws.ReactionMessageModifierNotification{}
+			if err := json.Unmarshal(msg.Content, notification); err != nil {
+				continue
+			}
+			if notification.IsExternalExtensions {
+				continue
+			}
+			if !notification.IsReact {
+				// first time to modify
+				var reactionExtensionList = make(map[string]unRelationTb.KeyValueModel)
+				extendMsg := unRelationTb.ExtendMsgModel{
+					ReactionExtensionList: reactionExtensionList,
+					ClientMsgID:           notification.ClientMsgID,
+					MsgFirstModifyTime:    notification.MsgFirstModifyTime,
+				}
+				for _, v := range notification.SuccessReactionExtensions {
+					reactionExtensionList[v.TypeKey] = unRelationTb.KeyValueModel{
+						TypeKey:          v.TypeKey,
+						Value:            v.Value,
+						LatestUpdateTime: v.LatestUpdateTime,
+					}
+				}
+
+				if err := mmc.extendMsgDatabase.InsertExtendMsg(ctx, notification.ConversationID, notification.SessionType, &extendMsg); err != nil {
+					// log.ZError(ctx, "MsgFirstModify InsertExtendMsg failed",  notification.ConversationID, notification.SessionType, extendMsg, err.Error())
+					continue
+				}
+			} else {
+				if err := mmc.extendMsgDatabase.InsertOrUpdateReactionExtendMsgSet(ctx, notification.ConversationID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, mmc.extendSetMsgModel.Pb2Model(notification.SuccessReactionExtensions)); err != nil {
+					// log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
+				}
+			}
+		} else if msg.ContentType == constant.ReactionMessageDeleter {
+			notification := &sdkws.ReactionMessageDeleteNotification{}
+			if err := json.Unmarshal(msg.Content, notification); err != nil {
+				continue
+			}
+			if err := mmc.extendMsgDatabase.DeleteReactionExtendMsgSet(ctx, notification.ConversationID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, mmc.extendSetMsgModel.Pb2Model(notification.SuccessReactionExtensions)); err != nil {
+				// log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
+			}
+		}
+	}
+}

internal/msgtransfer/online_history_msg_handler.go

@@ -0,0 +1,343 @@
+package msgtransfer
+
+import (
+	"context"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/kafka"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/rpcclient"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+	"github.com/Shopify/sarama"
+	"github.com/go-redis/redis"
+	"google.golang.org/protobuf/proto"
+)
+
+const ConsumerMsgs = 3
+const SourceMessages = 4
+const MongoMessages = 5
+const ChannelNum = 100
+
+type MsgChannelValue struct {
+	uniqueKey  string
+	ctx        context.Context
+	ctxMsgList []*ContextMsg
+}
+
+type TriggerChannelValue struct {
+	ctx      context.Context
+	cMsgList []*sarama.ConsumerMessage
+}
+
+type Cmd2Value struct {
+	Cmd   int
+	Value interface{}
+}
+type ContextMsg struct {
+	message *sdkws.MsgData
+	ctx     context.Context
+}
+
+type OnlineHistoryRedisConsumerHandler struct {
+	historyConsumerGroup *kafka.MConsumerGroup
+	chArrays             [ChannelNum]chan Cmd2Value
+	msgDistributionCh    chan Cmd2Value
+
+	singleMsgSuccessCount      uint64
+	singleMsgFailedCount       uint64
+	singleMsgSuccessCountMutex sync.Mutex
+	singleMsgFailedCountMutex  sync.Mutex
+
+	msgDatabase           controller.CommonMsgDatabase
+	conversationRpcClient *rpcclient.ConversationRpcClient
+	groupRpcClient        *rpcclient.GroupRpcClient
+}
+
+func NewOnlineHistoryRedisConsumerHandler(database controller.CommonMsgDatabase, conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient) *OnlineHistoryRedisConsumerHandler {
+	var och OnlineHistoryRedisConsumerHandler
+	och.msgDatabase = database
+	och.msgDistributionCh = make(chan Cmd2Value) //no buffer channel
+	go och.MessagesDistributionHandle()
+	for i := 0; i < ChannelNum; i++ {
+		och.chArrays[i] = make(chan Cmd2Value, 50)
+		go och.Run(i)
+	}
+	och.conversationRpcClient = conversationRpcClient
+	och.groupRpcClient = groupRpcClient
+	och.historyConsumerGroup = kafka.NewMConsumerGroup(&kafka.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
+		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.LatestMsgToRedis.Topic},
+		config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToRedis)
+	//statistics.NewStatistics(&och.singleMsgSuccessCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second singleMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
+	return &och
+}
+
+func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
+	for {
+		select {
+		case cmd := <-och.chArrays[channelID]:
+			switch cmd.Cmd {
+			case SourceMessages:
+				msgChannelValue := cmd.Value.(MsgChannelValue)
+				ctxMsgList := msgChannelValue.ctxMsgList
+				ctx := msgChannelValue.ctx
+				log.ZDebug(ctx, "msg arrived channel", "channel id", channelID, "msgList length", len(ctxMsgList), "uniqueKey", msgChannelValue.uniqueKey)
+				storageMsgList, notStorageMsgList, storageNotificationList, notStorageNotificationList, modifyMsgList := och.getPushStorageMsgList(ctxMsgList)
+				log.ZDebug(ctx, "msg lens", "storageMsgList", len(storageMsgList), "notStorageMsgList", len(notStorageMsgList),
+					"storageNotificationList", len(storageNotificationList), "notStorageNotificationList", len(notStorageNotificationList), "modifyMsgList", len(modifyMsgList))
+				conversationIDMsg := utils.GetChatConversationIDByMsg(ctxMsgList[0].message)
+				conversationIDNotification := utils.GetNotificationConversationID(ctxMsgList[0].message)
+				och.handleMsg(ctx, msgChannelValue.uniqueKey, conversationIDMsg, storageMsgList, notStorageMsgList)
+				och.handleNotification(ctx, msgChannelValue.uniqueKey, conversationIDNotification, storageNotificationList, notStorageNotificationList)
+				if err := och.msgDatabase.MsgToModifyMQ(ctx, msgChannelValue.uniqueKey, conversationIDNotification, modifyMsgList); err != nil {
+					log.ZError(ctx, "msg to modify mq error", err, "uniqueKey", msgChannelValue.uniqueKey, "modifyMsgList", modifyMsgList)
+				}
+			}
+		}
+	}
+}
+
+// 获取消息/通知 存储的消息列表， 不存储并且推送的消息列表，
+func (och *OnlineHistoryRedisConsumerHandler) getPushStorageMsgList(totalMsgs []*ContextMsg) (storageMsgList, notStorageMsgList, storageNotificatoinList, notStorageNotificationList, modifyMsgList []*sdkws.MsgData) {
+	isStorage := func(msg *sdkws.MsgData) bool {
+		options2 := utils.Options(msg.Options)
+		if options2.IsHistory() {
+			return true
+		} else {
+			// if !(!options2.IsSenderSync() && conversationID == msg.MsgData.SendID) {
+			// 	return false
+			// }
+			return false
+		}
+	}
+	for _, v := range totalMsgs {
+		options := utils.Options(v.message.Options)
+		if !options.IsNotNotification() {
+			// clone msg from notificationMsg
+			if options.IsSendMsg() {
+				msg := proto.Clone(v.message).(*sdkws.MsgData)
+				// 消息
+				if v.message.Options != nil {
+					msg.Options = utils.NewMsgOptions()
+				}
+				if options.IsOfflinePush() {
+					v.message.Options = utils.WithOptions(utils.Options(v.message.Options), utils.WithOfflinePush(false))
+					msg.Options = utils.WithOptions(utils.Options(msg.Options), utils.WithOfflinePush(true))
+				}
+				if options.IsUnreadCount() {
+					v.message.Options = utils.WithOptions(utils.Options(v.message.Options), utils.WithUnreadCount(false))
+					msg.Options = utils.WithOptions(utils.Options(msg.Options), utils.WithUnreadCount(true))
+				}
+				storageMsgList = append(storageMsgList, msg)
+			}
+			if isStorage(v.message) {
+				storageNotificatoinList = append(storageNotificatoinList, v.message)
+			} else {
+				notStorageNotificationList = append(notStorageNotificationList, v.message)
+			}
+		} else {
+			if isStorage(v.message) {
+				storageMsgList = append(storageMsgList, v.message)
+			} else {
+				notStorageMsgList = append(notStorageMsgList, v.message)
+			}
+		}
+		if v.message.ContentType == constant.ReactionMessageModifier || v.message.ContentType == constant.ReactionMessageDeleter {
+			modifyMsgList = append(modifyMsgList, v.message)
+		}
+	}
+	return
+}
+
+func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Context, key, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
+	och.toPushTopic(ctx, key, conversationID, notStorageList)
+	if len(storageList) > 0 {
+		lastSeq, _, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)
+		if err != nil {
+			log.ZError(ctx, "notification batch insert to redis error", err, "conversationID", conversationID, "storageList", storageList)
+			return
+		}
+		log.ZDebug(ctx, "success to next topic", "conversationID", conversationID)
+		och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+		och.toPushTopic(ctx, key, conversationID, storageList)
+	}
+}
+
+func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, key, conversationID string, msgs []*sdkws.MsgData) {
+	for _, v := range msgs {
+		och.msgDatabase.MsgToPushMQ(ctx, key, conversationID, v)
+	}
+}
+
+func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, key, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
+	och.toPushTopic(ctx, key, conversationID, notStorageList)
+	if len(storageList) > 0 {
+		lastSeq, isNewConversation, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)
+		if err != nil && errs.Unwrap(err) != redis.Nil {
+			log.ZError(ctx, "batch data insert to redis err", err, "storageMsgList", storageList)
+			och.singleMsgFailedCountMutex.Lock()
+			och.singleMsgFailedCount += uint64(len(storageList))
+			och.singleMsgFailedCountMutex.Unlock()
+			return
+		}
+		if isNewConversation {
+			if storageList[0].SessionType == constant.SuperGroupChatType {
+				log.ZInfo(ctx, "group chat first create conversation", "conversationID", conversationID)
+				userIDs, err := och.groupRpcClient.GetGroupMemberIDs(ctx, storageList[0].GroupID)
+				if err != nil {
+					log.ZWarn(ctx, "get group member ids error", err, "conversationID", conversationID)
+				} else {
+					if err := och.conversationRpcClient.GroupChatFirstCreateConversation(ctx, storageList[0].GroupID, userIDs); err != nil {
+						log.ZWarn(ctx, "single chat first create conversation error", err, "conversationID", conversationID)
+					}
+				}
+			} else {
+				if err := och.conversationRpcClient.SingleChatFirstCreateConversation(ctx, storageList[0].RecvID, storageList[0].SendID); err != nil {
+					log.ZWarn(ctx, "single chat first create conversation error", err, "conversationID", conversationID)
+				}
+			}
+		}
+
+		log.ZDebug(ctx, "success incr to next topic")
+		och.singleMsgSuccessCountMutex.Lock()
+		och.singleMsgSuccessCount += uint64(len(storageList))
+		och.singleMsgSuccessCountMutex.Unlock()
+		och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+		och.toPushTopic(ctx, key, conversationID, storageList)
+	}
+}
+
+func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
+	for {
+		aggregationMsgs := make(map[string][]*ContextMsg, ChannelNum)
+		select {
+		case cmd := <-och.msgDistributionCh:
+			switch cmd.Cmd {
+			case ConsumerMsgs:
+				triggerChannelValue := cmd.Value.(TriggerChannelValue)
+				ctx := triggerChannelValue.ctx
+				consumerMessages := triggerChannelValue.cMsgList
+				//Aggregation map[userid]message list
+				log.ZDebug(ctx, "batch messages come to distribution center", "length", len(consumerMessages))
+				for i := 0; i < len(consumerMessages); i++ {
+					ctxMsg := &ContextMsg{}
+					msgFromMQ := &sdkws.MsgData{}
+					err := proto.Unmarshal(consumerMessages[i].Value, msgFromMQ)
+					if err != nil {
+						log.ZError(ctx, "msg_transfer Unmarshal msg err", err, string(consumerMessages[i].Value))
+						continue
+					}
+					var arr []string
+					for i, header := range consumerMessages[i].Headers {
+						arr = append(arr, strconv.Itoa(i), string(header.Key), string(header.Value))
+					}
+					log.ZInfo(ctx, "consumer.kafka.GetContextWithMQHeader", "len", len(consumerMessages[i].Headers), "header", strings.Join(arr, ", "))
+					ctxMsg.ctx = kafka.GetContextWithMQHeader(consumerMessages[i].Headers)
+					ctxMsg.message = msgFromMQ
+					log.ZDebug(ctx, "single msg come to distribution center", "message", msgFromMQ, "key", string(consumerMessages[i].Key))
+					//aggregationMsgs[string(consumerMessages[i].Key)] = append(aggregationMsgs[string(consumerMessages[i].Key)], ctxMsg)
+					if oldM, ok := aggregationMsgs[string(consumerMessages[i].Key)]; ok {
+						oldM = append(oldM, ctxMsg)
+						aggregationMsgs[string(consumerMessages[i].Key)] = oldM
+					} else {
+						m := make([]*ContextMsg, 0, 100)
+						m = append(m, ctxMsg)
+						aggregationMsgs[string(consumerMessages[i].Key)] = m
+					}
+				}
+				log.ZDebug(ctx, "generate map list users len", "length", len(aggregationMsgs))
+				for uniqueKey, v := range aggregationMsgs {
+					if len(v) >= 0 {
+						hashCode := utils.GetHashCode(uniqueKey)
+						channelID := hashCode % ChannelNum
+						newCtx := withAggregationCtx(ctx, v)
+						log.ZDebug(newCtx, "generate channelID", "hashCode", hashCode, "channelID", channelID, "uniqueKey", uniqueKey)
+						och.chArrays[channelID] <- Cmd2Value{Cmd: SourceMessages, Value: MsgChannelValue{uniqueKey: uniqueKey, ctxMsgList: v, ctx: newCtx}}
+					}
+				}
+			}
+		}
+	}
+}
+func withAggregationCtx(ctx context.Context, values []*ContextMsg) context.Context {
+	var allMessageOperationID string
+	for i, v := range values {
+		if opid := mcontext.GetOperationID(v.ctx); opid != "" {
+			if i == 0 {
+				allMessageOperationID += opid
+
+			} else {
+				allMessageOperationID += "$" + opid
+			}
+		}
+	}
+	return mcontext.SetOperationID(ctx, allMessageOperationID)
+}
+
+func (och *OnlineHistoryRedisConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
+func (och *OnlineHistoryRedisConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error {
+	return nil
+}
+
+func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
+	for {
+		if sess == nil {
+			log.ZWarn(context.Background(), "sess == nil, waiting", nil)
+			time.Sleep(100 * time.Millisecond)
+		} else {
+			break
+		}
+	}
+	rwLock := new(sync.RWMutex)
+	log.ZDebug(context.Background(), "online new session msg come", "highWaterMarkOffset",
+		claim.HighWaterMarkOffset(), "topic", claim.Topic(), "partition", claim.Partition())
+	cMsg := make([]*sarama.ConsumerMessage, 0, 1000)
+	t := time.NewTicker(time.Millisecond * 100)
+	go func() {
+		for {
+			select {
+			case <-t.C:
+				if len(cMsg) > 0 {
+					rwLock.Lock()
+					ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
+					for _, v := range cMsg {
+						ccMsg = append(ccMsg, v)
+					}
+					cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
+					rwLock.Unlock()
+					split := 1000
+					ctx := mcontext.WithTriggerIDContext(context.Background(), utils.OperationIDGenerator())
+					log.ZDebug(ctx, "timer trigger msg consumer start", "length", len(ccMsg))
+					for i := 0; i < len(ccMsg)/split; i++ {
+						//log.Debug()
+						och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
+							ctx: ctx, cMsgList: ccMsg[i*split : (i+1)*split]}}
+					}
+					if (len(ccMsg) % split) > 0 {
+						och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
+							ctx: ctx, cMsgList: ccMsg[split*(len(ccMsg)/split):]}}
+					}
+					log.ZDebug(ctx, "timer trigger msg consumer end", "length", len(ccMsg))
+				}
+			}
+		}
+	}()
+	for msg := range claim.Messages() {
+		rwLock.Lock()
+		if len(msg.Value) != 0 {
+			cMsg = append(cMsg, msg)
+		}
+		rwLock.Unlock()
+		sess.MarkMessage(msg, "")
+	}
+	return nil
+}

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -0,0 +1,73 @@
+package msgtransfer
+
+import (
+	"context"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller"
+	kfk "github.com/OpenIMSDK/Open-IM-Server/pkg/common/kafka"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
+	"github.com/Shopify/sarama"
+	"google.golang.org/protobuf/proto"
+)
+
+type OnlineHistoryMongoConsumerHandler struct {
+	historyConsumerGroup *kfk.MConsumerGroup
+	msgDatabase          controller.CommonMsgDatabase
+}
+
+func NewOnlineHistoryMongoConsumerHandler(database controller.CommonMsgDatabase) *OnlineHistoryMongoConsumerHandler {
+	mc := &OnlineHistoryMongoConsumerHandler{
+		historyConsumerGroup: kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
+			OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.MsgToMongo.Topic},
+			config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMongo),
+		msgDatabase: database,
+	}
+	return mc
+}
+
+func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Context, cMsg *sarama.ConsumerMessage, key string, session sarama.ConsumerGroupSession) {
+	msg := cMsg.Value
+	msgFromMQ := pbMsg.MsgDataToMongoByMQ{}
+	err := proto.Unmarshal(msg, &msgFromMQ)
+	if err != nil {
+		log.ZError(ctx, "unmarshall failed", err, "key", key, "len", len(msg))
+		return
+	}
+	if len(msgFromMQ.MsgData) == 0 {
+		log.ZError(ctx, "msgFromMQ.MsgData is empty", nil, "cMsg", cMsg)
+		return
+	}
+	log.ZInfo(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.MsgData)
+	err = mc.msgDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq)
+	if err != nil {
+		log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
+	}
+	var seqs []int64
+	for _, msg := range msgFromMQ.MsgData {
+		seqs = append(seqs, msg.Seq)
+	}
+	err = mc.msgDatabase.DeleteMessagesFromCache(ctx, msgFromMQ.ConversationID, seqs)
+	if err != nil {
+		log.ZError(ctx, "remove cache msg from redis err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
+	}
+	mc.msgDatabase.DelUserDeleteMsgsList(ctx, msgFromMQ.ConversationID, seqs)
+}
+
+func (OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
+func (OnlineHistoryMongoConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
+func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
+	log.ZDebug(context.Background(), "online new session msg come", "highWaterMarkOffset",
+		claim.HighWaterMarkOffset(), "topic", claim.Topic(), "partition", claim.Partition())
+	for msg := range claim.Messages() {
+		ctx := mc.historyConsumerGroup.GetContextFromMsg(msg)
+		if len(msg.Value) != 0 {
+			mc.handleChatWs2Mongo(ctx, msg, string(msg.Key), sess)
+		} else {
+			log.ZError(ctx, "mongo msg get from kafka but is nil", nil, "conversationID", msg.Key)
+		}
+		sess.MarkMessage(msg, "")
+	}
+	return nil
+}

internal/msgtransfer/persistent_msg_handler.go

@@ -0,0 +1,88 @@
+/*
+** description("").
+** copyright('tuoyun,www.tuoyun.net').
+** author("fg,Gordon@tuoyun.net").
+** time(2021/5/11 15:37).
+ */
+package msgtransfer
+
+import (
+	"context"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller"
+	kfk "github.com/OpenIMSDK/Open-IM-Server/pkg/common/kafka"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+
+	"github.com/Shopify/sarama"
+	"google.golang.org/protobuf/proto"
+)
+
+type PersistentConsumerHandler struct {
+	persistentConsumerGroup *kfk.MConsumerGroup
+	chatLogDatabase         controller.ChatLogDatabase
+}
+
+func NewPersistentConsumerHandler(database controller.ChatLogDatabase) *PersistentConsumerHandler {
+	return &PersistentConsumerHandler{
+		persistentConsumerGroup: kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
+			OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.LatestMsgToRedis.Topic},
+			config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMySql),
+		chatLogDatabase: database,
+	}
+}
+
+func (pc *PersistentConsumerHandler) handleChatWs2Mysql(ctx context.Context, cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
+	msg := cMsg.Value
+	var tag bool
+	msgFromMQ := pbMsg.MsgDataToMQ{}
+	err := proto.Unmarshal(msg, &msgFromMQ)
+	if err != nil {
+		log.ZError(ctx, "msg_transfer Unmarshal msg err", err)
+		return
+	}
+	return
+	log.ZDebug(ctx, "handleChatWs2Mysql", "msg", msgFromMQ.MsgData)
+	//Control whether to store history messages (mysql)
+	isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
+	//Only process receiver data
+	if isPersist {
+		switch msgFromMQ.MsgData.SessionType {
+		case constant.SingleChatType, constant.NotificationChatType:
+			if msgKey == msgFromMQ.MsgData.RecvID {
+				tag = true
+			}
+		case constant.GroupChatType:
+			if msgKey == msgFromMQ.MsgData.SendID {
+				tag = true
+			}
+		case constant.SuperGroupChatType:
+			tag = true
+		}
+		if tag {
+			log.ZInfo(ctx, "msg_transfer msg persisting", "msg", string(msg))
+			if err = pc.chatLogDatabase.CreateChatLog(&msgFromMQ); err != nil {
+				log.ZError(ctx, "Message insert failed", err, "msg", msgFromMQ.String())
+				return
+			}
+		}
+	}
+}
+func (PersistentConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
+func (PersistentConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
+func (pc *PersistentConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	for msg := range claim.Messages() {
+		ctx := pc.persistentConsumerGroup.GetContextFromMsg(msg)
+		log.ZDebug(ctx, "kafka get info to mysql", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
+		if len(msg.Value) != 0 {
+			pc.handleChatWs2Mysql(ctx, msg, string(msg.Key), sess)
+		} else {
+			log.ZError(ctx, "msg get from kafka but is nil", nil, "key", msg.Key)
+		}
+		sess.MarkMessage(msg, "")
+	}
+	return nil
+}