|
|
@@ -346,11 +346,19 @@ func (ds *DataSync) syncclassdata(cifrom *dbo.ClassInfoHelper, fields, condition
|
|
|
return e
|
|
|
}
|
|
|
// 继续执行对象数据同步
|
|
|
+ interval := mcfg.GetDuration("datasync.run.interval", "1m")
|
|
|
+ lastsyncvtime = lastsyncvtime.Add(-interval)
|
|
|
+ if lastdatavtime.After(lastsyncvtime) {
|
|
|
+ lastsyncvtime = lastdatavtime
|
|
|
+ }
|
|
|
+ lastsyncvtime = lastsyncvtime.Add(-mcfg.GetDuration("datasync.run.overtime", "30s"))
|
|
|
nextvtime := lastsyncvtime
|
|
|
- for {
|
|
|
+ run := true
|
|
|
+ for run {
|
|
|
nextvtime = lastsyncvtime.Add(mcfg.GetDuration("datasync.pagetime", "1d"))
|
|
|
if time.Now().Before(nextvtime) {
|
|
|
- break
|
|
|
+ nextvtime = time.Now()
|
|
|
+ run = false
|
|
|
}
|
|
|
snextvtime := nextvtime.Format("2006-01-02 15:04:05")
|
|
|
mqlseg := mqlAddVtimeRange(mqlfrom, slastsyncvtime, snextvtime)
|
|
|
@@ -389,7 +397,7 @@ func (ds *DataSync) syncclassdata(cifrom *dbo.ClassInfoHelper, fields, condition
|
|
|
dc.LastSyncVtime = slastsyncvtime
|
|
|
ds.syncstatus.Save(mqlfrom, dc)
|
|
|
}
|
|
|
- logger.Info("sync data:", mqlfrom, "done", "vtime:", dc.FromVtime, "~", dc.LastSyncVtime, "records:", dc.RecordsCount, "lastvtime:", dc.LastDataVtime)
|
|
|
+ logger.Info("total sync data:", mqlfrom, "vtime:", dc.FromVtime, "~", dc.LastSyncVtime, "records:", dc.RecordsCount, "lastvtime:", dc.LastDataVtime)
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
@@ -522,12 +530,19 @@ func (ds *DataSync) syncbucketdata(mqlfrom string, cifrom, cito *dbo.ClassInfoHe
|
|
|
dc.LastSyncVtime = slastsyncvtime
|
|
|
}
|
|
|
// 继续执行数据同步
|
|
|
- count := 0
|
|
|
+ interval := mcfg.GetDuration("datasync.run.interval", "1m")
|
|
|
+ lastsyncvtime = lastsyncvtime.Add(-interval)
|
|
|
+ if lastdatavtime.After(lastsyncvtime) {
|
|
|
+ lastsyncvtime = lastdatavtime
|
|
|
+ }
|
|
|
+ lastsyncvtime = lastsyncvtime.Add(-mcfg.GetDuration("datasync.run.overtime", "30s"))
|
|
|
nextvtime := lastsyncvtime
|
|
|
- for {
|
|
|
+ run := true
|
|
|
+ for run {
|
|
|
nextvtime = lastsyncvtime.Add(mcfg.GetDuration("datasync.bucket.pagetime", "1h"))
|
|
|
if time.Now().Before(nextvtime) {
|
|
|
- break
|
|
|
+ nextvtime = time.Now()
|
|
|
+ run = false
|
|
|
}
|
|
|
snextvtime := nextvtime.Format("2006-01-02 15:04:05")
|
|
|
offset := 0
|
|
|
@@ -575,18 +590,13 @@ func (ds *DataSync) syncbucketdata(mqlfrom string, cifrom, cito *dbo.ClassInfoHe
|
|
|
lastsyncvtime = nextvtime
|
|
|
slastsyncvtime = lastsyncvtime.Format("2006-01-02 15:04:05")
|
|
|
recordscount += int64(offset)
|
|
|
- count += offset
|
|
|
//
|
|
|
dc.RecordsCount = recordscount
|
|
|
dc.LastDataVtime = lastdatavtime.Format("2006-01-02 15:04:05.000000")
|
|
|
dc.LastSyncVtime = slastsyncvtime
|
|
|
ds.syncstatus.Save(dckey, dc)
|
|
|
}
|
|
|
- logger.Debug("end sync", bucketType, "data", dc.BucketClass, dc.BucketField, "id:", dc.BucketObjID)
|
|
|
-
|
|
|
- if count > 0 {
|
|
|
- logger.Info("sync", bucketType, "data", dc.BucketClass, dc.BucketField, "id:", dc.BucketObjID, count, "records", "to time", dc.LastSyncVtime)
|
|
|
- }
|
|
|
+ logger.Info("total sync", bucketType, "data", dc.BucketClass, dc.BucketField, "id:", dc.BucketObjID, recordscount, "records", "to time", dc.LastSyncVtime)
|
|
|
return nil
|
|
|
}
|
|
|
|
