diff --git a/pkg/common/db/batch_insert_chat.go b/pkg/common/db/batch_insert_chat.go index b5fca3502..6a872529b 100644 --- a/pkg/common/db/batch_insert_chat.go +++ b/pkg/common/db/batch_insert_chat.go @@ -10,8 +10,134 @@ import ( "github.com/garyburd/redigo/redis" "github.com/golang/protobuf/proto" "go.mongodb.org/mongo-driver/bson" + "runtime" + "time" ) +func (d *DataBases) BatchInsertChat2DB(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string, currentMaxSeq uint64) error { + newTime := getCurrentTimestampByMill() + if len(msgList) > GetSingleGocMsgNum() { + return errors.New("too large") + } + isInit := false + var remain uint64 + blk0 := uint64(GetSingleGocMsgNum() - 1) + if currentMaxSeq < uint64(GetSingleGocMsgNum()) { + remain = blk0 - currentMaxSeq + } else { + excludeBlk0 := currentMaxSeq - blk0 + remain = (uint64(GetSingleGocMsgNum()) - (excludeBlk0 % uint64(GetSingleGocMsgNum()))) % uint64(GetSingleGocMsgNum()) + } + insertCounter := uint64(0) + msgListToMongo := make([]MsgInfo, 0) + msgListToMongoNext := make([]MsgInfo, 0) + seqUid := "" + seqUidNext := "" + log.Debug(operationID, "remain ", remain, "insertCounter ", insertCounter, "currentMaxSeq ", currentMaxSeq, userID, len(msgList)) + var err error + for _, m := range msgList { + log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID) + currentMaxSeq++ + sMsg := MsgInfo{} + sMsg.SendTime = m.MsgData.SendTime + m.MsgData.Seq = uint32(currentMaxSeq) + if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil { + return utils.Wrap(err, "") + } + if isInit { + msgListToMongoNext = append(msgListToMongoNext, sMsg) + seqUidNext = getSeqUid(userID, uint32(currentMaxSeq)) + log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain) + continue + } + if insertCounter < remain { + msgListToMongo = append(msgListToMongo, sMsg) + insertCounter++ + seqUid = getSeqUid(userID, uint32(currentMaxSeq)) + log.Debug(operationID, "msgListToMongo ", seqUid, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain) + } else { + msgListToMongoNext = append(msgListToMongoNext, sMsg) + seqUidNext = getSeqUid(userID, uint32(currentMaxSeq)) + log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain) + } + } + + ctx := context.Background() + c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat) + + if seqUid != "" { + filter := bson.M{"uid": seqUid} + log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo) + err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgListToMongo}}}).Err() + if err != nil { + log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter) + return utils.Wrap(err, "") + } + } + if seqUidNext != "" { + filter := bson.M{"uid": seqUidNext} + sChat := UserChat{} + sChat.UID = seqUidNext + sChat.Msg = msgListToMongoNext + log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext) + if _, err = c.InsertOne(ctx, &sChat); err != nil { + log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat) + return utils.Wrap(err, "") + } + } + log.NewWarn(operationID, "batch mgo cost time ", getCurrentTimestampByMill()-newTime, userID, len(msgList)) + return nil +} + +func (d *DataBases) BatchInsertChat2Cache(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) (error, uint64) { + newTime := getCurrentTimestampByMill() + if len(msgList) > GetSingleGocMsgNum() { + return errors.New("too large"), 0 + } + currentMaxSeq, err := d.GetUserMaxSeq(userID) + if err == nil { + + } else if err == redis.ErrNil { + currentMaxSeq = 0 + } else { + return utils.Wrap(err, ""), 0 + } + lastMaxSeq := currentMaxSeq + + for _, m := range msgList { + log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID) + currentMaxSeq++ + sMsg := MsgInfo{} + sMsg.SendTime = m.MsgData.SendTime + m.MsgData.Seq = uint32(currentMaxSeq) + } + log.Debug(operationID, "SetMessageToCache ", userID, len(msgList)) + err = d.SetMessageToCache(msgList, userID, operationID) + if err != nil { + log.Error(operationID, "setMessageToCache failed, continue ", err.Error(), len(msgList), userID) + } + log.NewWarn(operationID, "batch to redis cost time ", getCurrentTimestampByMill()-newTime, userID, len(msgList)) + return utils.Wrap(d.SetUserMaxSeq(userID, uint64(currentMaxSeq)), ""), lastMaxSeq +} + +func (d *DataBases) BatchInsertChatBoth(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) error { + err, lastMaxSeq := d.BatchInsertChat2Cache(userID, msgList, operationID) + if err != nil { + log.Error(operationID, "BatchInsertChat2Cache failed ", err.Error(), userID, len(msgList)) + return err + } + for { + if runtime.NumGoroutine() > 50000 { + log.NewWarn(operationID, "too many NumGoroutine ", runtime.NumGoroutine()) + time.Sleep(10 * time.Millisecond) + } else { + break + } + } + go d.BatchInsertChat2DB(userID, msgList, operationID, lastMaxSeq) + return nil +} + func (d *DataBases) BatchInsertChat(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) error { newTime := getCurrentTimestampByMill() if len(msgList) > GetSingleGocMsgNum() { @@ -104,3 +230,7 @@ func (d *DataBases) BatchInsertChat(userID string, msgList []*pbMsg.MsgDataToMQ, log.NewWarn(operationID, "batch mgo cost time ", getCurrentTimestampByMill()-newTime, userID, len(msgList)) return utils.Wrap(d.SetUserMaxSeq(userID, uint64(currentMaxSeq)), "") } + +//func (d *DataBases)setMessageToCache(msgList []*pbMsg.MsgDataToMQ, uid string) (err error) { +// +//}