diff --git a/business/jxstore/cms/sync.go b/business/jxstore/cms/sync.go index 2a696b7b3..04ebfae4e 100644 --- a/business/jxstore/cms/sync.go +++ b/business/jxstore/cms/sync.go @@ -402,35 +402,21 @@ func (v *VendorSync) SyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, vendo if isForce { dao.SetStoreSkuSyncStatus(db, loopMapInfo.VendorID, storeIDs, skuIDs, model.SyncFlagStoreSkuModifiedMask) } - if len(loopMapInfo.StoreMapList) > 1 { - var loopStoreTask tasksch.ITask - if model.MultiStoresVendorMap[loopMapInfo.VendorID] == 1 { - loopStoreTask = tasksch.NewSeqTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]), ctx, - 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 - } - } - return nil, err - }, 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) { - storeMap := batchItemList[0].(*model.StoreMap) - if _, err = handler.SyncStoreSkus(ctx, task, storeMap.StoreID, skuIDs, false, isContinueWhenError); err != nil { - globals.SugarLogger.Debugf("SyncStoresSkus failed2 store:%d failed with error:%v", storeMap.StoreID, err) - } - return nil, err - }, loopMapInfo.StoreMapList) - } - t.AddChild(loopStoreTask).Run() - _, err = loopStoreTask.GetResult(0) - } else { - _, err = handler.SyncStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, skuIDs, false, isContinueWhenError) + parallelCount := 5 + if model.MultiStoresVendorMap[loopMapInfo.VendorID] == 1 { + parallelCount = 1 } + loopStoreTask := tasksch.NewParallelTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]), + tasksch.NewParallelConfig().SetParallelCount(parallelCount).SetIsContinueWhenError(isContinueWhenError), ctx, + func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { + storeMap := batchItemList[0].(*model.StoreMap) + if _, err = handler.SyncStoreSkus(ctx, task, storeMap.StoreID, skuIDs, false, isContinueWhenError); err != nil { + globals.SugarLogger.Debugf("SyncStoresSkus failed2 store:%d failed with error:%v", storeMap.StoreID, err) + } + return nil, err + }, loopMapInfo.StoreMapList) + t.AddChild(loopStoreTask).Run() + _, err = loopStoreTask.GetResult(0) } return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID) }, isContinueWhenError) @@ -458,35 +444,21 @@ func (v *VendorSync) FullSyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, v func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) { loopMapInfo := batchItemList[0].(*LoopStoreMapInfo) if handler := v.GetStoreHandler(loopMapInfo.VendorID); handler != nil { - if len(loopMapInfo.StoreMapList) > 1 { - var loopStoreTask tasksch.ITask - if model.MultiStoresVendorMap[loopMapInfo.VendorID] == 1 { - loopStoreTask = tasksch.NewSeqTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]), ctx, - 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 - } - } - return nil, err - }, 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) { - storeMap := batchItemList[0].(*model.StoreMap) - if _, err = handler.FullSyncStoreSkus(ctx, task, storeMap.StoreID, false, isContinueWhenError); err != nil { - globals.SugarLogger.Debugf("FullSyncStoresSkus failed2 store:%d failed with error:%v", storeMap.StoreID, err) - } - return nil, err - }, loopMapInfo.StoreMapList) - } - t.AddChild(loopStoreTask).Run() - _, err = loopStoreTask.GetResult(0) - } else { - _, err = handler.FullSyncStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, false, isContinueWhenError) + parallelCount := 5 + if model.MultiStoresVendorMap[loopMapInfo.VendorID] == 1 { + parallelCount = 1 } + loopStoreTask := tasksch.NewParallelTask(fmt.Sprintf("处理平台%s", model.VendorChineseNames[loopMapInfo.VendorID]), + tasksch.NewParallelConfig().SetParallelCount(parallelCount).SetIsContinueWhenError(isContinueWhenError), ctx, + func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { + storeMap := batchItemList[0].(*model.StoreMap) + if _, err = handler.FullSyncStoreSkus(ctx, task, storeMap.StoreID, false, isContinueWhenError); err != nil { + globals.SugarLogger.Debugf("FullSyncStoresSkus failed2 store:%d failed with error:%v", storeMap.StoreID, err) + } + return nil, err + }, loopMapInfo.StoreMapList) + t.AddChild(loopStoreTask).Run() + _, err = loopStoreTask.GetResult(0) } return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID) }, isContinueWhenError)