111 lines
2.3 KiB
Go
111 lines
2.3 KiB
Go
package statutils
|
|
|
|
import (
|
|
"stat/common/utils"
|
|
"stat/models/syncrecord"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
type SyncItem struct {
|
|
AppID int32 // 应用ID
|
|
JobName string // 任务名称
|
|
JobDesc string // 任务描述
|
|
today time.Time // 内部日期
|
|
todayCount int64 // 今日拉取条数
|
|
CurrentCount int64 // 当前拉取条数
|
|
RequestBegin time.Time // 请求开始时间
|
|
RequestEnd time.Time // 请求结束时间
|
|
ProcessBegin time.Time // 处理开始时间
|
|
ProcessEnd time.Time // 处理结束时间
|
|
jobBegin time.Time // 任务开始时间
|
|
jobEnd time.Time // 任务结束时间
|
|
DataLatest time.Time // 拉到的数据的最后时间
|
|
Error error // 失败错误
|
|
load bool // 是否载入redis数据
|
|
}
|
|
|
|
func (s *SyncItem) IncCount(value int64) {
|
|
s.todayCount += value
|
|
s.CurrentCount += value
|
|
}
|
|
|
|
func (s *SyncItem) TodayCount() int64 {
|
|
return s.todayCount
|
|
}
|
|
|
|
func (s *SyncItem) JobBegin() time.Time {
|
|
return s.jobBegin
|
|
}
|
|
|
|
func (s *SyncItem) SetJobBegin(start time.Time) {
|
|
today := utils.BeginningOfDay(start)
|
|
s.jobBegin = start
|
|
if !s.load {
|
|
s.load = true
|
|
s.today = today
|
|
GetTodayCount(s)
|
|
}
|
|
if !s.today.Equal(today) {
|
|
s.today = today
|
|
s.todayCount = 0
|
|
s.CurrentCount = 0
|
|
}
|
|
}
|
|
|
|
func (s *SyncItem) JobEnd() time.Time {
|
|
return s.jobEnd
|
|
}
|
|
|
|
func (s *SyncItem) SetJobEnd(end time.Time) {
|
|
s.jobEnd = end
|
|
SetTodayCount(s)
|
|
}
|
|
|
|
type SyncMonitor struct {
|
|
mu sync.Mutex
|
|
dict map[int32]map[string]*SyncItem
|
|
}
|
|
|
|
func (s *SyncMonitor) SetData(appId int32, job string, setFunc func(item *SyncItem)) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
jobs := s.dict[appId]
|
|
if jobs == nil {
|
|
jobs = map[string]*SyncItem{}
|
|
s.dict[appId] = jobs
|
|
}
|
|
item := jobs[job]
|
|
if item == nil {
|
|
item = &SyncItem{AppID: appId, JobName: job, JobDesc: syncrecord.SyncJobDescribe[job]}
|
|
jobs[job] = item
|
|
}
|
|
item.AppID = appId
|
|
item.JobName = job
|
|
|
|
setFunc(item)
|
|
|
|
// protect some item attr
|
|
item.AppID = appId
|
|
item.JobName = job
|
|
item.JobDesc = syncrecord.SyncJobDescribe[job]
|
|
}
|
|
|
|
func (s *SyncMonitor) Traverse(fn func(item *SyncItem)) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
for _, jobs := range s.dict {
|
|
for _, item := range jobs {
|
|
fn(item)
|
|
}
|
|
}
|
|
}
|
|
|
|
var (
|
|
GlobalSyncMonitor = &SyncMonitor{
|
|
dict: map[int32]map[string]*SyncItem{},
|
|
}
|
|
)
|