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

- 京东门店商品信息同步，部分失败后尝试单个同步

Browse Source
This commit is contained in:
gazebo
2019-06-13 11:43:46 +08:00
parent 8ce9457c4b
commit 19908aee23
1 changed files with 103 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
business/partner/purchase/jd/store_sku.go
View File

@@ -33,16 +33,17 @@ func (p *PurchaseHandler) syncStoreSkus(ctx *jxcontext.Context, parentTask tasks
return "", err return "", err
} }
batchSize := jdapi.MaxStoreSkuBatchSize batchSize := jdapi.MaxStoreSkuBatchSize
storeSkusLen := len(storeSkus) // storeSkusLen := len(storeSkus)
if storeSkusLen < jdapi.MaxStoreSkuBatchSize*2 { // if storeSkusLen < jdapi.MaxStoreSkuBatchSize/2 {
batchSize = (storeSkusLen + 2) / 3 // batchSize = 1
} else if storeSkusLen < jdapi.MaxStoreSkuBatchSize { // } else if storeSkusLen < jdapi.MaxStoreSkuBatchSize {
batchSize = (storeSkusLen + 1) / 2 // batchSize = (storeSkusLen + 1) / 2
} else if storeSkusLen < jdapi.MaxStoreSkuBatchSize/2 { // } else if storeSkusLen < jdapi.MaxStoreSkuBatchSize*2 {
batchSize = 1 // batchSize = (storeSkusLen + 2) / 3
} // }
task := tasksch.NewParallelTask("syncStoreSkus京东", tasksch.NewParallelConfig().SetBatchSize(batchSize).SetIsContinueWhenError(isContinueWhenError), ctx, task := tasksch.NewParallelTask("syncStoreSkus京东", tasksch.NewParallelConfig().SetBatchSize(batchSize).SetIsContinueWhenError(isContinueWhenError), ctx,
func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
doWork := func(batchItemList []interface{}) (err error) {
var skuPriceInfoList []*jdapi.SkuPriceInfo var skuPriceInfoList []*jdapi.SkuPriceInfo
var skuVendibilityList []*jdapi.StockVendibility var skuVendibilityList []*jdapi.StockVendibility
var skuStockList []*jdapi.SkuStock var skuStockList []*jdapi.SkuStock
@@ -126,7 +127,7 @@ func (p *PurchaseHandler) syncStoreSkus(ctx *jxcontext.Context, parentTask tasks
syncMask = -1 syncMask = -1
} }
if syncMask != 0 && len(batchBindIDs) > 0 { if syncMask != 0 && len(batchBindIDs) > 0 {
db := dao.GetDB() // 多线程问题 // db := dao.GetDB() // 多线程问题
sql := ` sql := `
UPDATE store_sku_bind t1 UPDATE store_sku_bind t1
SET t1.jd_sync_status = t1.jd_sync_status & ? SET t1.jd_sync_status = t1.jd_sync_status & ?
@@ -140,6 +141,14 @@ func (p *PurchaseHandler) syncStoreSkus(ctx *jxcontext.Context, parentTask tasks
} else if len(errList) > 1 { } else if len(errList) > 1 {
err = fmt.Errorf("%v", errList) err = fmt.Errorf("%v", errList)
} }
return err
}
err = doWork(batchItemList)
if isErrPartialFailed(err) && len(batchItemList) > 1 {
for _, v := range batchItemList {
doWork([]interface{}{v})
}
}
return nil, err return nil, err
}, storeSkus) }, storeSkus)
tasksch.HandleTask(task, parentTask, false).Run() tasksch.HandleTask(task, parentTask, false).Run()
@@ -149,6 +158,13 @@ func (p *PurchaseHandler) syncStoreSkus(ctx *jxcontext.Context, parentTask tasks
return task.ID, err return task.ID, err
} }
func isErrPartialFailed(err error) bool {
if errExt, ok := err.(*utils.ErrorWithCode); ok && errExt.Code() == jdapi.ResponseInnerCodePartialFailed {
return true
}
return false
}
func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, skuIDs []int, isAsync, isContinueWhenError bool) (hint string, err error) { func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, skuIDs []int, isAsync, isContinueWhenError bool) (hint string, err error) {
globals.SugarLogger.Debugf("jd SyncStoresSkus, storeID:%d, skuIDs:%v", storeID, skuIDs) globals.SugarLogger.Debugf("jd SyncStoresSkus, storeID:%d, skuIDs:%v", storeID, skuIDs)
db := dao.GetDB() db := dao.GetDB()