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

fix delete

Browse Source
This commit is contained in:
wangchuxiao 2022-08-10 12:02:50 +08:00
parent 6c0922a6d8
commit a74168d1e1
4 changed files with 64 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
config/config.yaml
View File

@ -32,6 +32,7 @@ mongo:
dbPassword: #mongo密码，建议先不设置 dbPassword: #mongo密码，建议先不设置
dbMaxPoolSize: 100 dbMaxPoolSize: 100
dbRetainChatRecords: 3650 #mongo保存离线消息时间（天），根据需求修改 dbRetainChatRecords: 3650 #mongo保存离线消息时间（天），根据需求修改
chatRecordsClearTime: "* * * * *" # 每天凌晨3点清除消息，该配置和linux定时任务一样， 清理操作建议设置在用户活跃少的时候 # 0 3 * * *
redis: redis:
dbAddress: [ 127.0.0.1:16379 ] #redis地址 单机时，填写一个地址即可，使用redis集群时候，填写集群中多个节点地址（主从地址都可以填写，增加容灾能力），默认即可 dbAddress: [ 127.0.0.1:16379 ] #redis地址 单机时，填写一个地址即可，使用redis集群时候，填写集群中多个节点地址（主从地址都可以填写，增加容灾能力），默认即可

69
internal/cron_task/clear_msg.go
View File

@ -7,47 +7,72 @@ import (
server_api_params "Open_IM/pkg/proto/sdk_ws" server_api_params "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"strconv"
"strings"
) )
const oldestList = 0 const oldestList = 0
const newestList = -1 const newestList = -1
func ResetUserGroupMinSeq(operationID, groupID, userID string) error { func ResetUserGroupMinSeq(operationID, groupID string, userIDList []string) error {
var delMsgIDList [][2]interface{}
minSeq, err := deleteMongoMsg(operationID, groupID, oldestList, &delMsgIDList)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), groupID, "deleteMongoMsg failed")
return utils.Wrap(err, "")
}
for _, userID := range userIDList {
userMinSeq, err := db.DB.GetGroupUserMinSeq(groupID, userID)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "GetGroupUserMinSeq failed", groupID, userID, err.Error())
continue
}
if userMinSeq > uint64(minSeq) {
err = db.DB.SetGroupUserMinSeq(groupID, userID, userMinSeq)
} else {
err = db.DB.SetGroupUserMinSeq(groupID, userID, uint64(minSeq))
}
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userID, userMinSeq, minSeq)
}
}
return nil return nil
} }
func DeleteMongoMsgAndResetRedisSeq(operationID, userID string) error { func DeleteMongoMsgAndResetRedisSeq(operationID, userID string) error {
// -1 表示从当前最早的一个开始 var delMsgIDList [][2]interface{}
var delMsgIDList []string
minSeq, err := deleteMongoMsg(operationID, userID, oldestList, &delMsgIDList) minSeq, err := deleteMongoMsg(operationID, userID, oldestList, &delMsgIDList)
if err != nil { if err != nil {
return utils.Wrap(err, "") return utils.Wrap(err, "")
} }
log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDList: ", delMsgIDList) log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDMap: ", userID, delMsgIDList)
err = db.DB.SetUserMinSeq(userID, minSeq) err = db.DB.SetUserMinSeq(userID, minSeq)
return err return err
} }
// recursion // recursion
func deleteMongoMsg(operationID string, ID string, index int64, IDList *[]string) (uint32, error) { func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[][2]interface{}) (uint32, error) {
// 从最旧的列表开始找 // 从最旧的列表开始找
msgs, err := db.DB.GetUserMsgListByIndex(ID, index) msgs, err := db.DB.GetUserMsgListByIndex(ID, index)
if err != nil { if err != nil {
return 0, utils.Wrap(err, "GetUserMsgListByIndex failed") return 0, utils.Wrap(err, "GetUserMsgListByIndex failed")
} }
if len(msgs.Msg) > db.GetSingleGocMsgNum() {
log.NewWarn(operationID, utils.GetSelfFuncName(), "msgs too large", len(msgs.Msg), msgs.UID)
}
log.NewDebug(operationID, utils.GetSelfFuncName(), "get msgs: ", msgs.UID) log.NewDebug(operationID, utils.GetSelfFuncName(), "get msgs: ", msgs.UID)
for i, msg := range msgs.Msg { for i, msg := range msgs.Msg {
// 找到列表中不需要删除的消息了 // 找到列表中不需要删除的消息了
if msg.SendTime+int64(config.Config.Mongo.DBRetainChatRecords) > utils.GetCurrentTimestampByMill() { if utils.GetCurrentTimestampByMill() < msg.SendTime+int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000 {
if len(*IDList) > 0 { if len(*delMsgIDList) > 0 {
err := db.DB.DelMongoMsgs(*IDList) var IDList []string
for _, v := range *delMsgIDList {
IDList = append(IDList, v[0].(string))
}
err := db.DB.DelMongoMsgs(IDList)
if err != nil { if err != nil {
return 0, utils.Wrap(err, "DelMongoMsgs failed") return 0, utils.Wrap(err, "DelMongoMsgs failed")
} }
} }
minSeq := getDelMaxSeqByIDList(*IDList) minSeq := getDelMaxSeqByIDList(*delMsgIDList)
if i > 0 { if i > 0 {
msgPb := &server_api_params.MsgData{} msgPb := &server_api_params.MsgData{}
err = proto.Unmarshal(msg.Msg, msgPb) err = proto.Unmarshal(msg.Msg, msgPb)
@ -65,24 +90,24 @@ func deleteMongoMsg(operationID string, ID string, index int64, IDList *[]string
return minSeq, nil return minSeq, nil
} }
} }
*IDList = append(*IDList, msgs.UID) msgPb := &server_api_params.MsgData{}
err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, msgPb)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID)
return 0, utils.Wrap(err, "proto.Unmarshal failed")
}
*delMsgIDList = append(*delMsgIDList, [2]interface{}{msgs.UID, msgPb.Seq})
// 没有找到 代表需要全部删除掉 继续递归查找下一个比较旧的列表 // 没有找到 代表需要全部删除掉 继续递归查找下一个比较旧的列表
seq, err := deleteMongoMsg(operationID, utils.GetSelfFuncName(), index-1, IDList) seq, err := deleteMongoMsg(operationID, utils.GetSelfFuncName(), index+1, delMsgIDList)
if err != nil { if err != nil {
return 0, utils.Wrap(err, "deleteMongoMsg failed") return 0, utils.Wrap(err, "deleteMongoMsg failed")
} }
return seq, nil return seq, nil
} }
func getDelMaxSeqByIDList(IDList []string) uint32 { func getDelMaxSeqByIDList(delMsgIDList [][2]interface{}) uint32 {
if len(IDList) == 0 { if len(delMsgIDList) == 0 {
return 0 return 0
} }
l := strings.Split(IDList[len(IDList)-1], ":") return delMsgIDList[len(delMsgIDList)-1][1].(uint32)
index, _ := strconv.Atoi(l[len(l)-1])
if index == 0 {
// 4999
return uint32(db.GetSingleGocMsgNum()) - 1
} // 5000
return (uint32(db.GetSingleGocMsgNum()) - 1) + uint32(index*db.GetSingleGocMsgNum())
} }

9
internal/cron_task/cron_task.go
View File

@ -6,6 +6,7 @@ import (
rocksCache "Open_IM/pkg/common/db/rocks_cache" rocksCache "Open_IM/pkg/common/db/rocks_cache"
"Open_IM/pkg/common/log" "Open_IM/pkg/common/log"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"fmt"
"github.com/robfig/cron/v3" "github.com/robfig/cron/v3"
) )
@ -37,11 +38,10 @@ func StartCronTask() {
continue continue
} }
log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "userIDList:", userIDList) log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "userIDList:", userIDList)
for _, userID := range userIDList { if err := ResetUserGroupMinSeq(operationID, groupID, userIDList); err != nil {
if err := ResetUserGroupMinSeq(operationID, groupID, userID); err != nil { log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userIDList)
log.NewError(operationID, utils.GetSelfFuncName(), operationID, groupID, userID, err.Error())
}
} }
} }
} else { } else {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error()) log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
@ -49,6 +49,7 @@ func StartCronTask() {
} }
}) })
if err != nil { if err != nil {
fmt.Println("start cron failed", err.Error())
panic(err) panic(err)
} }
c.Start() c.Start()

21
pkg/common/config/config.go
View File

@ -92,16 +92,17 @@ type config struct {
DBMaxLifeTime int `yaml:"dbMaxLifeTime"` DBMaxLifeTime int `yaml:"dbMaxLifeTime"`
} }
Mongo struct { Mongo struct {
DBUri string `yaml:"dbUri"` DBUri string `yaml:"dbUri"`
DBAddress string `yaml:"dbAddress"` DBAddress string `yaml:"dbAddress"`
DBDirect bool `yaml:"dbDirect"` DBDirect bool `yaml:"dbDirect"`
DBTimeout int `yaml:"dbTimeout"` DBTimeout int `yaml:"dbTimeout"`
DBDatabase string `yaml:"dbDatabase"` DBDatabase string `yaml:"dbDatabase"`
DBSource string `yaml:"dbSource"` DBSource string `yaml:"dbSource"`
DBUserName string `yaml:"dbUserName"` DBUserName string `yaml:"dbUserName"`
DBPassword string `yaml:"dbPassword"` DBPassword string `yaml:"dbPassword"`
DBMaxPoolSize int `yaml:"dbMaxPoolSize"` DBMaxPoolSize int `yaml:"dbMaxPoolSize"`
DBRetainChatRecords int `yaml:"dbRetainChatRecords"` DBRetainChatRecords int `yaml:"dbRetainChatRecords"`
ChatRecordsClearTime string `yaml:"chatRecordsClearTime"`
} }
Redis struct { Redis struct {
DBAddress []string `yaml:"dbAddress"` DBAddress []string `yaml:"dbAddress"`