Files
51dj_server/skd/job/event/event.go
T
2026-10-08 16:50:13 +08:00

120 lines
2.5 KiB
Go

package event
import (
"91jd-server/common/dataReport"
"91jd-server/common/log"
"91jd-server/models/v/eventmod"
"encoding/json"
"go.mongodb.org/mongo-driver/bson"
)
var isRuining = false
const (
maxRetryCount = 10 // 最大重试次数
maxReportNum = 100 // 最大上报数
)
func Report() {
//防止重入
if isRuining {
return
}
isRuining = true
defer func() {
isRuining = false
}()
// 查询出一千条记录批量上传
list, err := eventmod.GetList(bson.M{
"$or": []bson.M{
bson.M{
"status": eventmod.Default,
},
bson.M{
"status": eventmod.Failure,
"retryCount": bson.M{"$lte": maxRetryCount},
},
},
}, 1000)
if err != nil {
log.Error("Get event list error", log.E(err))
return
}
if len(list) == 0 {
return
}
var eventList []*dataReport.CommonField
for _, v := range list {
event := &dataReport.CommonField{}
err = json.Unmarshal([]byte(v.Data), &event)
if err != nil {
log.Error("event report json.Unmarshal fail", log.Any("eventData", v.Data), log.E(err))
continue
}
eventList = append(eventList, event)
if len(eventList) < maxReportNum {
continue
}
// 批量上报
err = dataReport.BatchReport(eventList)
if err != nil {
log.Error("Batch report error", log.E(err))
// 将这些设置为失败
onReportEventFail(eventList)
} else {
onReportEventSuccess(eventList)
}
eventList = []*dataReport.CommonField{} // 重新变成0个
}
// 将剩余的批量上报
err = dataReport.BatchReport(eventList)
if err != nil {
log.Error("Batch report error", log.E(err))
// 将这些设置为失败
onReportEventFail(eventList)
} else {
onReportEventSuccess(eventList)
}
}
func onReportEventSuccess(eventList []*dataReport.CommonField) {
eventIds := []string{}
for _, v := range eventList {
eventIds = append(eventIds, v.EventID)
}
if len(eventIds) == 0 {
return
}
err := eventmod.UpdateMany(bson.M{
"eventId": bson.M{"$in": eventIds},
}, bson.M{
"$set": bson.M{"status": eventmod.Success},
})
if err != nil {
log.Error("Update event fail", log.E(err))
return
}
}
func onReportEventFail(eventList []*dataReport.CommonField) {
eventIds := []string{}
for _, v := range eventList {
eventIds = append(eventIds, v.EventID)
}
if len(eventIds) == 0 {
return
}
err := eventmod.UpdateMany(bson.M{
"eventId": bson.M{"$in": eventIds},
}, bson.M{
"$set": bson.M{"status": eventmod.Failure},
"$inc": bson.M{"retryCount": 1},
})
if err != nil {
log.Error("Update event fail", log.E(err))
return
}
}