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

cron

Browse Source
This commit is contained in:
wangchuxiao 2022-11-24 19:17:29 +08:00
parent 9be70d640e
commit 0654ecc9bb
3 changed files with 77 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cmd/open_im_cron_task/main.go
View File

@ -2,11 +2,15 @@ package main
import ( import (
"Open_IM/internal/cron_task" "Open_IM/internal/cron_task"
"flag"
"fmt" "fmt"
"time" "time"
) )
func main() { func main() {
fmt.Println(time.Now(), "start cronTask") var userID = flag.String("userID", "", "userID to clear msg and reset seq")
cronTask.StartCronTask() var workingGroupID = flag.String("workingGroupID", "", "workingGroupID to clear msg and reset seq")
flag.Parse()
fmt.Println(time.Now(), "start cronTask", userID, workingGroupID)
cronTask.StartCronTask(*userID, *workingGroupID)
} }

1
internal/cron_task/clear_msg.go
View File

@ -139,6 +139,7 @@ func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMs
} }
delStruct.minSeq = lastMsgPb.Seq delStruct.minSeq = lastMsgPb.Seq
if msgListIsFull(msgs) { if msgListIsFull(msgs) {
log.NewDebug(operationID, "msg list is full", msgs.UID)
delStruct.delUidList = append(delStruct.delUidList, msgs.UID) delStruct.delUidList = append(delStruct.delUidList, msgs.UID)
} }
log.NewDebug(operationID, ID, "continue", delStruct) log.NewDebug(operationID, ID, "continue", delStruct)

108
internal/cron_task/cron_task.go
View File

@ -14,60 +14,31 @@ import (
const cronTaskOperationID = "cronTaskOperationID-" const cronTaskOperationID = "cronTaskOperationID-"
func StartCronTask() { func StartCronTask(userID, workingGroupID string) {
log.NewPrivateLog("cron") log.NewPrivateLog("cron")
log.NewInfo(utils.OperationIDGenerator(), "start cron task", "cron config", config.Config.Mongo.ChatRecordsClearTime) log.NewInfo(utils.OperationIDGenerator(), "start cron task", "cron config", config.Config.Mongo.ChatRecordsClearTime)
c := cron.New()
fmt.Println("cron config", config.Config.Mongo.ChatRecordsClearTime) fmt.Println("cron config", config.Config.Mongo.ChatRecordsClearTime)
_, err := c.AddFunc(config.Config.Mongo.ChatRecordsClearTime, func() { if userID != "" {
// user msg clear
operationID := getCronTaskOperationID() operationID := getCronTaskOperationID()
log.NewInfo(operationID, "====================== start del cron task ======================") StartClearMsg(operationID, []string{userID})
userIDList, err := im_mysql_model.SelectAllUserID()
if err == nil {
log.NewDebug(operationID, utils.GetSelfFuncName(), "userIDList: ", userIDList)
userIDList = []string{"4158779020"}
for _, userID := range userIDList {
if err := DeleteMongoMsgAndResetRedisSeq(operationID, userID); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), userID)
} }
if err := checkMaxSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil { if workingGroupID != "" {
log.NewError(operationID, utils.GetSelfFuncName(), userID, err) operationID := getCronTaskOperationID()
StartClearWorkingGroupMsg(operationID, []string{workingGroupID})
} }
if userID != "" || workingGroupID != "" {
fmt.Println("clear msg finished")
return
} }
} else { clearFunc := func() {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error()) ClearAll()
} }
//return c := cron.New()
// working group msg clear _, err := c.AddFunc(config.Config.Mongo.ChatRecordsClearTime, clearFunc)
workingGroupIDList, err := im_mysql_model.GetGroupIDListByGroupType(constant.WorkingGroup)
if err == nil {
log.NewDebug(operationID, utils.GetSelfFuncName(), "workingGroupIDList: ", workingGroupIDList)
for _, groupID := range workingGroupIDList {
userIDList, err = rocksCache.GetGroupMemberIDListFromCache(groupID)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID)
continue
}
log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "userIDList:", userIDList)
if err := ResetUserGroupMinSeq(operationID, groupID, userIDList); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userIDList)
}
if err := checkMaxSeqWithMongo(operationID, groupID, constant.ReadDiffusion); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), groupID, err)
}
}
} else {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
}
log.NewInfo(operationID, "====================== start del cron finished ======================")
})
if err != nil { if err != nil {
fmt.Println("start cron failed", err.Error(), config.Config.Mongo.ChatRecordsClearTime) fmt.Println("start cron failed", err.Error(), config.Config.Mongo.ChatRecordsClearTime)
panic(err) panic(err)
} }
c.Start() c.Start()
fmt.Println("start cron task success") fmt.Println("start cron task success")
for { for {
@ -78,3 +49,56 @@ func StartCronTask() {
func getCronTaskOperationID() string { func getCronTaskOperationID() string {
return cronTaskOperationID + utils.OperationIDGenerator() return cronTaskOperationID + utils.OperationIDGenerator()
} }
func ClearAll() {
operationID := getCronTaskOperationID()
log.NewInfo(operationID, "====================== start del cron task ======================")
//var userIDList []string
var err error
userIDList, err := im_mysql_model.SelectAllUserID()
if err == nil {
StartClearMsg(operationID, userIDList)
} else {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
}
//return
// working group msg clear
workingGroupIDList, err := im_mysql_model.GetGroupIDListByGroupType(constant.WorkingGroup)
if err == nil {
StartClearWorkingGroupMsg(operationID, workingGroupIDList)
} else {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
}
log.NewInfo(operationID, "====================== start del cron finished ======================")
}
func StartClearMsg(operationID string, userIDList []string) {
log.NewDebug(operationID, utils.GetSelfFuncName(), "userIDList: ", userIDList)
for _, userID := range userIDList {
if err := DeleteMongoMsgAndResetRedisSeq(operationID, userID); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), userID)
}
if err := checkMaxSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), userID, err)
}
}
}
func StartClearWorkingGroupMsg(operationID string, workingGroupIDList []string) {
log.NewDebug(operationID, utils.GetSelfFuncName(), "workingGroupIDList: ", workingGroupIDList)
for _, groupID := range workingGroupIDList {
userIDList, err := rocksCache.GetGroupMemberIDListFromCache(groupID)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID)
continue
}
log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "workingGroupIDList:", userIDList)
if err := ResetUserGroupMinSeq(operationID, groupID, userIDList); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userIDList)
}
if err := checkMaxSeqWithMongo(operationID, groupID, constant.ReadDiffusion); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), groupID, err)
}
}
}