liulei/jx-callback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- 减少SyncStoresSkus与FullSyncStoresSkus的冗余部分

Browse Source
This commit is contained in:
gazebo
2019-07-23 22:02:15 +08:00
parent 643a1c1037
commit db215e2891
1 changed files with 28 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
business/jxstore/cms/sync.go
View File

@@ -402,22 +402,12 @@ func (v *VendorSync) SyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, vendo
if isForce { if isForce {
dao.SetStoreSkuSyncStatus(db, loopMapInfo.VendorID, storeIDs, skuIDs, model.SyncFlagStoreSkuModifiedMask) dao.SetStoreSkuSyncStatus(db, loopMapInfo.VendorID, storeIDs, skuIDs, model.SyncFlagStoreSkuModifiedMask)
} }
if len(loopMapInfo.StoreMapList) > 1 { parallelCount := 5
var loopStoreTask tasksch.ITask
if model.MultiStoresVendorMap[loopMapInfo.VendorID] == 1 { if model.MultiStoresVendorMap[loopMapInfo.VendorID] == 1 {
loopStoreTask = tasksch.NewSeqTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]), ctx, parallelCount = 1
func(task *tasksch.SeqTask, step int, params ...interface{}) (result interface{}, err error) {
storeID := loopMapInfo.StoreMapList[step].StoreID
if _, err = handler.SyncStoreSkus(ctx, task, storeID, skuIDs, false, isContinueWhenError); err != nil {
globals.SugarLogger.Debugf("SyncStoresSkus failed1 store:%d failed with error:%v", storeID, err)
if isContinueWhenError {
err = nil
} }
} loopStoreTask := tasksch.NewParallelTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]),
return nil, err tasksch.NewParallelConfig().SetParallelCount(parallelCount).SetIsContinueWhenError(isContinueWhenError), ctx,
}, len(loopMapInfo.StoreMapList))
} else {
loopStoreTask = tasksch.NewParallelTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]), tasksch.NewParallelConfig().SetIsContinueWhenError(isContinueWhenError), ctx,
func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
storeMap := batchItemList[0].(*model.StoreMap) storeMap := batchItemList[0].(*model.StoreMap)
if _, err = handler.SyncStoreSkus(ctx, task, storeMap.StoreID, skuIDs, false, isContinueWhenError); err != nil { if _, err = handler.SyncStoreSkus(ctx, task, storeMap.StoreID, skuIDs, false, isContinueWhenError); err != nil {
@@ -425,12 +415,8 @@ func (v *VendorSync) SyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, vendo
} }
return nil, err return nil, err
}, loopMapInfo.StoreMapList) }, loopMapInfo.StoreMapList)
}
t.AddChild(loopStoreTask).Run() t.AddChild(loopStoreTask).Run()
_, err = loopStoreTask.GetResult(0) _, err = loopStoreTask.GetResult(0)
} else {
_, err = handler.SyncStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, skuIDs, false, isContinueWhenError)
}
} }
return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID) return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}, isContinueWhenError) }, isContinueWhenError)
@@ -458,22 +444,12 @@ func (v *VendorSync) FullSyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, v
func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) { func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) {
loopMapInfo := batchItemList[0].(*LoopStoreMapInfo) loopMapInfo := batchItemList[0].(*LoopStoreMapInfo)
if handler := v.GetStoreHandler(loopMapInfo.VendorID); handler != nil { if handler := v.GetStoreHandler(loopMapInfo.VendorID); handler != nil {
if len(loopMapInfo.StoreMapList) > 1 { parallelCount := 5
var loopStoreTask tasksch.ITask
if model.MultiStoresVendorMap[loopMapInfo.VendorID] == 1 { if model.MultiStoresVendorMap[loopMapInfo.VendorID] == 1 {
loopStoreTask = tasksch.NewSeqTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]), ctx, parallelCount = 1
func(task *tasksch.SeqTask, step int, params ...interface{}) (result interface{}, err error) {
storeID := loopMapInfo.StoreMapList[step].StoreID
if _, err = handler.FullSyncStoreSkus(ctx, task, storeID, false, isContinueWhenError); err != nil {
globals.SugarLogger.Debugf("FullSyncStoresSkus failed1 store:%d failed with error:%v", storeID, err)
if isContinueWhenError {
err = nil
} }
} loopStoreTask := tasksch.NewParallelTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]),
return nil, err tasksch.NewParallelConfig().SetParallelCount(parallelCount).SetIsContinueWhenError(isContinueWhenError), ctx,
}, len(loopMapInfo.StoreMapList))
} else {
loopStoreTask = tasksch.NewParallelTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]), tasksch.NewParallelConfig().SetIsContinueWhenError(isContinueWhenError), ctx,
func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
storeMap := batchItemList[0].(*model.StoreMap) storeMap := batchItemList[0].(*model.StoreMap)
if _, err = handler.FullSyncStoreSkus(ctx, task, storeMap.StoreID, false, isContinueWhenError); err != nil { if _, err = handler.FullSyncStoreSkus(ctx, task, storeMap.StoreID, false, isContinueWhenError); err != nil {
@@ -481,12 +457,8 @@ func (v *VendorSync) FullSyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, v
} }
return nil, err return nil, err
}, loopMapInfo.StoreMapList) }, loopMapInfo.StoreMapList)
}
t.AddChild(loopStoreTask).Run() t.AddChild(loopStoreTask).Run()
_, err = loopStoreTask.GetResult(0) _, err = loopStoreTask.GetResult(0)
} else {
_, err = handler.FullSyncStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, false, isContinueWhenError)
}
} }
return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID) return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}, isContinueWhenError) }, isContinueWhenError)