From 8768fe82363d02ffe2d00274acd7d43d22f14438 Mon Sep 17 00:00:00 2001 From: gazebo Date: Wed, 7 Aug 2019 14:15:30 +0800 Subject: [PATCH] =?UTF-8?q?-=20FreeBatchStoreSkuInfo=E4=B8=8EFreeBatchStor?= =?UTF-8?q?eSkuSyncInfo=E6=B7=BB=E5=8A=A0=E4=BB=BB=E5=8A=A1=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- business/jxstore/cms/sync_store_sku.go | 18 ++++++++++-------- business/jxstore/misc/misc2.go | 2 +- business/jxutils/tasksch/parallel_task.go | 4 +--- business/jxutils/tasksch/sequence_task.go | 4 +--- business/jxutils/tasksch/task.go | 15 +++++++++++++-- business/partner/purchase/jd/store_sku2.go | 2 +- business/partner/putils/store_sku.go | 12 ++++++------ 7 files changed, 33 insertions(+), 24 deletions(-) diff --git a/business/jxstore/cms/sync_store_sku.go b/business/jxstore/cms/sync_store_sku.go index 3428e9280..08f2cb326 100644 --- a/business/jxstore/cms/sync_store_sku.go +++ b/business/jxstore/cms/sync_store_sku.go @@ -349,7 +349,7 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, isFull bo switch step { case 0: if len(deleteList) > 0 { - _, err = putils.FreeBatchStoreSkuInfo(func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { + _, err = putils.FreeBatchStoreSkuInfo("删除门店商品", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { var successList []*partner.StoreSkuInfo if successList, err = singleStoreHandler.DeleteStoreSkus(ctx, storeID, vendorStoreID, batchedStoreSkuList); singleStoreHandler.IsErrSkuNotExist(err) { err = nil @@ -365,7 +365,7 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, isFull bo } case 1: if len(createList) > 0 { - _, err = putils.FreeBatchStoreSkuSyncInfo(func(task tasksch.ITask, batchedStoreSkuList []*dao.StoreSkuSyncInfo) (result interface{}, successCount int, err error) { + _, err = putils.FreeBatchStoreSkuSyncInfo("创建门店商品", func(task tasksch.ITask, batchedStoreSkuList []*dao.StoreSkuSyncInfo) (result interface{}, successCount int, err error) { var successList []*dao.StoreSkuSyncInfo if successList, err = singleStoreHandler.CreateStoreSkus(ctx, storeID, vendorStoreID, batchedStoreSkuList); singleStoreHandler.IsErrSkuExist(err) { if skuNameList, err2 := singleStoreHandler.GetStoreSkusFullInfo(ctx, task, storeID, vendorStoreID, []*partner.StoreSkuInfo{ @@ -392,7 +392,7 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, isFull bo } case 2: if len(updateList) > 0 { - _, err = putils.FreeBatchStoreSkuSyncInfo(func(task tasksch.ITask, batchedStoreSkuList []*dao.StoreSkuSyncInfo) (result interface{}, successCount int, err error) { + _, err = putils.FreeBatchStoreSkuSyncInfo("更新门店商品基础信息", func(task tasksch.ITask, batchedStoreSkuList []*dao.StoreSkuSyncInfo) (result interface{}, successCount int, err error) { var successList []*dao.StoreSkuSyncInfo if successList, err = singleStoreHandler.UpdateStoreSkus(ctx, storeID, vendorStoreID, batchedStoreSkuList); err == nil { successList = batchedStoreSkuList @@ -406,7 +406,7 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, isFull bo case 3: for k, list := range [][]*partner.StoreSkuInfo{stockList /*, onlineList*/} { if len(list) > 0 { - _, err = putils.FreeBatchStoreSkuInfo(func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { + _, err = putils.FreeBatchStoreSkuInfo("更新门店商品库存", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { var successList []*partner.StoreSkuInfo if successList, err = storeSkuHandler.UpdateStoreSkusStock(ctx, storeID, vendorStoreID, batchedStoreSkuList); err == nil { successList = batchedStoreSkuList @@ -421,12 +421,14 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, isFull bo case 4, 5: statusList := onlineList status := model.SkuStatusNormal + name := "可售门店商品" if step == 5 { statusList = offlineList status = model.SkuStatusDontSale + name = "不可售门店商品" } if len(statusList) > 0 { - _, err = putils.FreeBatchStoreSkuInfo(func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { + _, err = putils.FreeBatchStoreSkuInfo(name, func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { var successList []*partner.StoreSkuInfo if successList, err = storeSkuHandler.UpdateStoreSkusStatus(ctx, storeID, vendorStoreID, batchedStoreSkuList, status); err == nil { successList = batchedStoreSkuList @@ -439,7 +441,7 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, isFull bo } case 6: if len(priceList) > 0 { - _, err = putils.FreeBatchStoreSkuInfo(func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { + _, err = putils.FreeBatchStoreSkuInfo("更新门店商品价格", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { var successList []*partner.StoreSkuInfo if successList, err = storeSkuHandler.UpdateStoreSkusPrice(ctx, storeID, vendorStoreID, batchedStoreSkuList); err == nil { successList = batchedStoreSkuList @@ -473,7 +475,7 @@ func PruneMissingStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, ven localSkuMap[v.SkuID] = v } var sku2Delete []*partner.StoreSkuInfo - task := tasksch.NewSeqTask(fmt.Sprintf("PruneMissingStoreSkus平台:%s", model.VendorChineseNames[vendorID]), ctx, + task := tasksch.NewSeqTask(fmt.Sprintf("清除平台:%s上多余的门店商品", model.VendorChineseNames[vendorID]), ctx, func(task *tasksch.SeqTask, step int, params ...interface{}) (result interface{}, err error) { switch step { case 0: @@ -490,7 +492,7 @@ func PruneMissingStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, ven } case 1: if len(sku2Delete) > 0 { - _, err = putils.FreeBatchStoreSkuInfo(func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { + _, err = putils.FreeBatchStoreSkuInfo("删除门店商品", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { _, err = handler.DeleteStoreSkus(ctx, storeID, vendorStoreID, sku2Delete) return nil, 0, err }, ctx, parentTask, sku2Delete, handler.GetStoreSkusBatchSize(partner.FuncDeleteStoreSkus), isContinueWhenError) diff --git a/business/jxstore/misc/misc2.go b/business/jxstore/misc/misc2.go index b55a57367..716b7f4cb 100644 --- a/business/jxstore/misc/misc2.go +++ b/business/jxstore/misc/misc2.go @@ -116,7 +116,7 @@ func StartOrEndOpStore(isStart bool, startTime, endTime int16, isAsync, isContin AddOrDelExtraStoreOptime(vendorID, storeID, vendorStoreID, &storeListValue.Store, startOpStoreTime, endOpStoreTime, false) } - _, err = putils.FreeBatchStoreSkuInfo(func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { + _, err = putils.FreeBatchStoreSkuInfo("更新门店商品库存", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { for _, skuValue := range batchedStoreSkuList { skuValue.Stock = GetStockValue(isStart) } diff --git a/business/jxutils/tasksch/parallel_task.go b/business/jxutils/tasksch/parallel_task.go index e5e85c121..de62f3cf8 100644 --- a/business/jxutils/tasksch/parallel_task.go +++ b/business/jxutils/tasksch/parallel_task.go @@ -130,9 +130,7 @@ func (task *ParallelTask) Run() { if err != nil { // 出错 // globals.SugarLogger.Infof("ParallelTask.Run %s, subtask(job:%s, params:%s) result:%v, failed with error:%v", task.Name, utils.Format4Output(job, true), utils.Format4Output(task.params, true), result, err) if task.IsContinueWhenError { - task.locker.Lock() - task.batchErrList = append(task.batchErrList, err) - task.locker.Unlock() + task.AddBatchErr(err) } else { chanRetVal = err goto end diff --git a/business/jxutils/tasksch/sequence_task.go b/business/jxutils/tasksch/sequence_task.go index 38cab7cf8..c443c85e6 100644 --- a/business/jxutils/tasksch/sequence_task.go +++ b/business/jxutils/tasksch/sequence_task.go @@ -44,9 +44,7 @@ func (task *SeqTask) Run() { if err != nil { // globals.SugarLogger.Infof("SeqTask.Run %s step:%d failed with error:%v", task.Name, i, err) if task.IsContinueWhenError { - task.locker.Lock() - task.batchErrList = append(task.batchErrList, err) - task.locker.Unlock() + task.AddBatchErr(err) } else { taskErr = err break diff --git a/business/jxutils/tasksch/task.go b/business/jxutils/tasksch/task.go index a364aa4fd..9228a6be1 100644 --- a/business/jxutils/tasksch/task.go +++ b/business/jxutils/tasksch/task.go @@ -3,6 +3,7 @@ package tasksch import ( "encoding/json" "fmt" + "strings" "sync" "time" @@ -283,6 +284,14 @@ func (t *BaseTask) getResult() []interface{} { return t.Result } +func (t *BaseTask) AddBatchErr(err error) { + if err != nil { + t.locker.Lock() + defer t.locker.Unlock() + t.batchErrList = append(t.batchErrList, err) + } +} + // func (t *BaseTask) GetOriginalErr() error { // t.locker.RLock() // defer t.locker.RUnlock() @@ -330,9 +339,11 @@ func (t *BaseTask) Error() (errMsg string) { errMsg += "," + t.mainErr.Error() } else { errMsg += fmt.Sprintf("部分失败, 总共:%d, 成功:%d, 失败:%d, 详情如下:\n", t.TotalItemCount, t.FinishedItemCount, t.FailedItemCount) - for _, v := range t.batchErrList { - errMsg += fmt.Sprintf("%s,\n", v.Error()) + strList := make([]string, len(t.batchErrList)) + for k, v := range t.batchErrList { + strList[k] = v.Error() } + errMsg += strings.Join(strList, ",\n") } t.locker.Lock() t.Err = errMsg diff --git a/business/partner/purchase/jd/store_sku2.go b/business/partner/purchase/jd/store_sku2.go index d5c7e140e..ec790bd9f 100644 --- a/business/partner/purchase/jd/store_sku2.go +++ b/business/partner/purchase/jd/store_sku2.go @@ -65,7 +65,7 @@ func (p *PurchaseHandler) getStoreSkusBareInfoLimitSize(ctx *jxcontext.Context, } func (p *PurchaseHandler) GetStoreSkusBareInfo(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, vendorStoreID string, inStoreSkuList []*partner.StoreSkuInfo) (outStoreSkuList []*partner.StoreSkuInfo, err error) { - result, err := putils.FreeBatchStoreSkuInfo(func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { + result, err := putils.FreeBatchStoreSkuInfo("获取门店商品信息", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { list, err := p.getStoreSkusBareInfoLimitSize(ctx, task, storeID, vendorStoreID, batchedStoreSkuList) if err == nil { result = list diff --git a/business/partner/putils/store_sku.go b/business/partner/putils/store_sku.go index 57da5d120..652d519b1 100644 --- a/business/partner/putils/store_sku.go +++ b/business/partner/putils/store_sku.go @@ -29,7 +29,7 @@ func (p *DefSingleStorePlatform) DeleteStoreAllSkus(ctx *jxcontext.Context, pare VendorSkuID: v.SkuList[0].VendorSkuID, } } - _, err = FreeBatchStoreSkuInfo(func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { + _, err = FreeBatchStoreSkuInfo("删除门店商品", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { _, err = p.DeleteStoreSkus(ctx, storeID, vendorStoreID, batchedStoreSkuList) return nil, 0, err }, ctx, parentTask, storeStoreList, p.GetStoreSkusBatchSize(partner.FuncDeleteStoreSkus), isContinueWhenError) @@ -76,7 +76,7 @@ func flatCatList(catList []*partner.BareCategoryInfo) (flattedCatList []*partner } func (p *DefSingleStorePlatform) GetStoreSkusBareInfo(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, vendorStoreID string, inStoreSkuList []*partner.StoreSkuInfo) (outStoreSkuList []*partner.StoreSkuInfo, err error) { - resultList, err := FreeBatchStoreSkuInfo(func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { + resultList, err := FreeBatchStoreSkuInfo("获取门店商品信息", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) { result, err = p.GetStoreSkusFullInfo(ctx, parentTask, storeID, vendorStoreID, batchedStoreSkuList) return result, successCount, err }, ctx, parentTask, inStoreSkuList, p.GetStoreSkusBatchSize(partner.FuncGetStoreSkusFullInfo), true) @@ -134,9 +134,9 @@ func (p *DefSingleStorePlatform) GetStoreCategory(ctx *jxcontext.Context, storeI return cat, err } -func FreeBatchStoreSkuInfo(handler func(tasksch.ITask, []*partner.StoreSkuInfo) (interface{}, int, error), ctx *jxcontext.Context, parentTask tasksch.ITask, storeSkuList []*partner.StoreSkuInfo, batchSize int, isContinueWhenError bool) (resultList []interface{}, err error) { +func FreeBatchStoreSkuInfo(name string, handler func(tasksch.ITask, []*partner.StoreSkuInfo) (interface{}, int, error), ctx *jxcontext.Context, parentTask tasksch.ITask, storeSkuList []*partner.StoreSkuInfo, batchSize int, isContinueWhenError bool) (resultList []interface{}, err error) { if true { //len(storeSkuList) > batchSize { - task := tasksch.NewParallelTask2("FreeBatchStoreSkuInfo", tasksch.NewParallelConfig().SetParallelCount(1).SetBatchSize(batchSize).SetIsContinueWhenError(isContinueWhenError), ctx, + task := tasksch.NewParallelTask2(fmt.Sprintf("FreeBatchStoreSkuInfo:%s", name), tasksch.NewParallelConfig().SetParallelCount(1).SetBatchSize(batchSize).SetIsContinueWhenError(isContinueWhenError), ctx, func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, successCount int, err error) { batchStoreSkuList := make([]*partner.StoreSkuInfo, len(batchItemList)) for k, v := range batchItemList { @@ -159,9 +159,9 @@ func FreeBatchStoreSkuInfo(handler func(tasksch.ITask, []*partner.StoreSkuInfo) return resultList, err } -func FreeBatchStoreSkuSyncInfo(handler func(tasksch.ITask, []*dao.StoreSkuSyncInfo) (interface{}, int, error), ctx *jxcontext.Context, parentTask tasksch.ITask, storeSkuList []*dao.StoreSkuSyncInfo, batchSize int, isContinueWhenError bool) (resultList []interface{}, err error) { +func FreeBatchStoreSkuSyncInfo(name string, handler func(tasksch.ITask, []*dao.StoreSkuSyncInfo) (interface{}, int, error), ctx *jxcontext.Context, parentTask tasksch.ITask, storeSkuList []*dao.StoreSkuSyncInfo, batchSize int, isContinueWhenError bool) (resultList []interface{}, err error) { if true { //len(storeSkuList) > batchSize { - task := tasksch.NewParallelTask2("FreeBatchStoreSkuSyncInfo", tasksch.NewParallelConfig().SetParallelCount(1).SetBatchSize(batchSize).SetIsContinueWhenError(isContinueWhenError), ctx, + task := tasksch.NewParallelTask2(fmt.Sprintf("FreeBatchStoreSkuSyncInfo:%s", name), tasksch.NewParallelConfig().SetParallelCount(1).SetBatchSize(batchSize).SetIsContinueWhenError(isContinueWhenError), ctx, func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, successCount int, err error) { batchStoreSkuList := make([]*dao.StoreSkuSyncInfo, len(batchItemList)) for k, v := range batchItemList {