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

- 重构tasksch

Browse Source
This commit is contained in:
gazebo
2019-08-01 16:20:52 +08:00
parent 66fcd42fc6
commit 852fff6eff
5 changed files with 131 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,7 +4,6 @@ import (
"errors"
"fmt"
"reflect"
"strings"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils"
@@ -421,19 +420,19 @@ func (v *VendorSync) SyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, vendo
return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}, isContinueWhenError)
if task != nil {
if vendorErr := partner.IsErrChangePriceFailed(task.GetOriginalErr()); vendorErr != nil {
platformList := make([]string, len(task.GetDetailErrList()))
for k, v := range task.GetDetailErrList() {
if vendorErr := partner.IsErrVendorError(v); vendorErr != nil {
platformList[k] = model.VendorChineseNames[vendorErr.VendorID()]
} else {
platformList[k] = "未知"
}
}
err = fmt.Errorf("同步价格失败\n失败平台：%s", strings.Join(platformList, ","))
} else {
err = makeSyncError(err)
}
// if vendorErr := partner.IsErrChangePriceFailed(task.GetOriginalErr()); vendorErr != nil {
// platformList := make([]string, len(task.GetDetailErrList()))
// for k, v := range task.GetDetailErrList() {
// if vendorErr := partner.IsErrVendorError(v); vendorErr != nil {
// platformList[k] = model.VendorChineseNames[vendorErr.VendorID()]
// } else {
// platformList[k] = "未知"
// }
// }
// err = fmt.Errorf("同步价格失败\n失败平台：%s", strings.Join(platformList, ","))
// } else {
// }
err = makeSyncError(err)
}
return hint, err
}

23
business/jxutils/tasksch/parallel_task.go
View File

@@ -2,7 +2,6 @@ package tasksch
import (
"errors"
"time"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils"
@@ -167,29 +166,9 @@ func (task *ParallelTask) Run() {
}
}
task.locker.Lock()
if taskErr != nil { // 如果有错误，肯定就是失败了
task.Status = TaskStatusFailed
} else {
if task.FinishedJobCount+task.FailedJobCount < task.TotalJobCount {
taskErr = ErrTaskIsCanceled
task.Status = TaskStatusCanceled
} else {
task.Status = TaskStatusFinished
}
}
if taskErr != nil {
task.OriginalErr = taskErr
task.Err = NewTaskError(task.Name, taskErr)
} else {
if len(task.batchErrList) > 0 {
task.OriginalErr = task.batchErrList[0]
}
task.Err = task.buildTaskErrFromBatchErrList()
}
task.Result = taskResult
task.TerminatedAt = time.Now()
task.mainErr = taskErr
task.locker.Unlock()
globals.SugarLogger.Debugf("ParallelTask.Run %s, err:%v", task.Name, task.Err)
close(task.subFinishChan)
task.jobList = nil // 如果不释放，任务被管理的话，会导致内存不能释放

24
business/jxutils/tasksch/sequence_task.go
View File

@@ -1,8 +1,6 @@
package tasksch
import (
"time"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
"git.rosy.net.cn/jx-callback/globals"
@@ -57,29 +55,9 @@ func (task *SeqTask) Run() {
}
EndFor:
task.locker.Lock()
if taskErr != nil { // 如果有错误，肯定就是失败了
task.Status = TaskStatusFailed
} else {
if task.FinishedJobCount+task.FailedJobCount < task.TotalJobCount {
taskErr = ErrTaskIsCanceled
task.Status = TaskStatusCanceled
} else {
task.Status = TaskStatusFinished
}
}
if taskErr != nil {
task.OriginalErr = taskErr
task.Err = NewTaskError(task.Name, taskErr)
} else {
if len(task.batchErrList) > 0 {
task.OriginalErr = task.batchErrList[0]
}
task.Err = task.buildTaskErrFromBatchErrList()
}
task.Result = taskResult
task.TerminatedAt = time.Now()
task.mainErr = taskErr
task.locker.Unlock()
globals.SugarLogger.Debugf("SeqTask.Run %s, result:%v, err:%v", task.Name, taskResult, task.Err)
})
}

165
business/jxutils/tasksch/task.go
View File

@@ -3,7 +3,6 @@ package tasksch
import (
"encoding/json"
"fmt"
"strings"
"sync"
"time"
@@ -58,35 +57,37 @@ type ITask interface {
AddChild(task ITask) ITask
GetChildren() TaskList
SetParent(parentTask ITask)
GetOriginalErr() error
GetDetailErrList() []error
// GetOriginalErr() error
GetErr() error
// GetDetailErrList() []error
GetLeafResult() (finishedItemCount, failedItemCount int)
json.Marshaler
}
type TaskError struct {
name string
errStr string
}
// type TaskError struct {
// name string
// errStr string
// }
func (t *TaskError) MarshalJSON() ([]byte, error) {
return json.Marshal(t.Error())
}
// func (t *TaskError) MarshalJSON() ([]byte, error) {
// return json.Marshal(t.Error())
// }
func (t *TaskError) Error() string {
return fmt.Sprintf("[%s], 错误:%s", t.name, t.errStr)
}
// func (t *TaskError) Error() string {
// return fmt.Sprintf("[%s], 错误:%s", t.name, t.errStr)
// }
func (t *TaskError) String() string {
return t.Error()
}
// func (t *TaskError) String() string {
// return t.Error()
// }
func NewTaskError(name string, err error) *TaskError {
return &TaskError{
name: name,
errStr: err.Error(),
}
}
// func NewTaskError(name string, err error) *TaskError {
// return &TaskError{
// name: name,
// errStr: err.Error(),
// }
// }
type BaseTask struct {
Name string `json:"name"`
@@ -108,11 +109,11 @@ type BaseTask struct {
NoticeMsg string `json:"noticeMsg"`
Result []interface{} `json:"-"`
Children TaskList `json:"children"`
Err error `json:"err"`
OriginalErr error `json:"-"`
Result []interface{} `json:"-"`
Children TaskList `json:"children"`
Err string `json:"err"`
mainErr error
batchErrList []error
finishChan chan struct{}
@@ -168,7 +169,7 @@ func (t *BaseTask) GetID() string {
func (t *BaseTask) GetResult(duration time.Duration) (retVal []interface{}, err error) {
if t.GetStatus() >= TaskStatusEndBegin {
return t.getResult(), t.GetOriginalErr()
return t.getResult(), t.GetErr()
}
if duration == 0 {
duration = time.Hour * 10000 // duration为0表示无限等待
@@ -178,7 +179,7 @@ func (t *BaseTask) GetResult(duration time.Duration) (retVal []interface{}, err
case <-t.finishChan:
t.isGetResultCalled = true
timer.Stop()
return t.getResult(), t.GetOriginalErr()
return t.getResult(), t.GetErr()
case <-timer.C:
}
return nil, ErrTaskNotFinished
@@ -282,24 +283,69 @@ func (t *BaseTask) getResult() []interface{} {
return t.Result
}
// func (t *BaseTask) GetOriginalErr() error {
// t.locker.RLock()
// defer t.locker.RUnlock()
// return nil
// }
func (t *BaseTask) GetErr() error {
t.locker.RLock()
defer t.locker.RUnlock()
return t.Err
if t.mainErr == nil && len(t.batchErrList) == 0 {
return nil
}
return t
}
func (t *BaseTask) GetOriginalErr() error {
t.locker.RLock()
defer t.locker.RUnlock()
return t.OriginalErr
func (t *BaseTask) GetLeafResult() (finishedItemCount, failedItemCount int) {
if len(t.Children) == 0 {
return t.FinishedItemCount, t.FailedItemCount
}
for _, v := range t.Children {
subFinishedItemCount, subFailedItemCount := v.GetLeafResult()
finishedItemCount += subFinishedItemCount
failedItemCount += subFailedItemCount
}
return finishedItemCount, failedItemCount
}
func (t *BaseTask) GetDetailErrList() []error {
func (t *BaseTask) Error() (errMsg string) {
if t.mainErr == nil && len(t.batchErrList) == 0 {
return ""
}
t.locker.RLock()
defer t.locker.RUnlock()
return t.batchErrList
errMsg = t.Err
t.locker.RUnlock()
if errMsg != "" {
return errMsg
}
errMsg = "任务:" + t.Name
if t.parent == nil {
finishedItemCount, failedItemCount := t.GetLeafResult()
errMsg += fmt.Sprintf(", 全部总共:%d, 成功:%d, 失败:%d,\n", (finishedItemCount + failedItemCount), finishedItemCount, failedItemCount)
}
if t.mainErr != nil {
errMsg += "失败"
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())
}
}
t.locker.Lock()
t.Err = errMsg
t.locker.Unlock()
return errMsg
}
// func (t *BaseTask) GetDetailErrList() []error {
// t.locker.RLock()
// defer t.locker.RUnlock()
// return t.batchErrList
// }
func AddChild(parentTask ITask, task ITask) ITask {
if parentTask != nil {
return parentTask.AddChild(task)
@@ -334,6 +380,25 @@ func (t *BaseTask) run(taskHandler func()) {
}()
taskHandler()
task := t
task.locker.Lock()
if task.mainErr != nil { // 如果有错误，肯定就是失败了
task.Status = TaskStatusFailed
} else {
if task.FinishedJobCount+task.FailedJobCount < task.TotalJobCount {
task.mainErr = ErrTaskIsCanceled
task.Status = TaskStatusCanceled
} else {
task.Status = TaskStatusFinished
}
}
task.TerminatedAt = time.Now()
task.locker.Unlock()
task.Error()
globals.SugarLogger.Debugf("Task:%s, result:%v, err:%v", task.Name, task.Result, task.mainErr)
select {
case <-t.quitChan:
default:
@@ -353,18 +418,14 @@ func (t *BaseTask) run(taskHandler func()) {
if authInfo, err := t.ctx.GetV2AuthInfo(); err == nil { // 这里应该是不管登录类型，直接以可能的方式发消息
var content string
taskDesc := fmt.Sprintf("你的异步任务[%s],ID[%s],开始于:%s,结束于:%s,", t.Name, t.ID, utils.Time2Str(t.CreatedAt), utils.Time2Str(t.TerminatedAt))
if t.Err == nil {
if t.mainErr == nil {
content = fmt.Sprintf("%s执行%s", taskDesc, TaskStatusName[t.Status])
noticeMsg := t.GetNoticeMsg()
if noticeMsg != "" {
content += ",通知消息:" + noticeMsg
}
} else {
if t.Status == TaskStatusFinished {
content = fmt.Sprintf("%s执行部分失败,%s", taskDesc, t.Err.Error())
} else {
content = fmt.Sprintf("%s执行失败,%s", taskDesc, t.Err.Error())
}
content = t.Error()
}
msg.SendUserMessage(dingdingapi.MsgTyeText, authInfo.UserID, "异步任务完成", content)
}
@@ -394,16 +455,16 @@ func (t *BaseTask) setStatus(status int) {
t.Status = status
}
func (t *BaseTask) buildTaskErrFromBatchErrList() (err error) {
if len(t.batchErrList) > 0 {
strList := make([]string, len(t.batchErrList))
for k, v := range t.batchErrList {
strList[k] = v.Error()
}
return NewTaskError(t.Name, fmt.Errorf("总共:%d, 成功:%d, 失败:%d, 详情:\n%s", t.TotalItemCount, t.FinishedItemCount, t.FailedItemCount, strings.Join(strList, "\n")))
}
return nil
}
// func (t *BaseTask) buildTaskErrFromBatchErrList() (err error) {
// if len(t.batchErrList) > 0 {
// strList := make([]string, len(t.batchErrList))
// for k, v := range t.batchErrList {
// strList[k] = v.Error()
// }
// return NewTaskError(t.Name, fmt.Errorf("总共:%d, 成功:%d, 失败:%d, 详情:\n%s", t.TotalItemCount, t.FinishedItemCount, t.FailedItemCount, strings.Join(strList, "\n")))
// }
// return nil
// }
func (task *BaseTask) callWorker(worker func() (retVal interface{}, err error)) (retVal interface{}, err error) {
defer func() {

10
business/jxutils/tasksch/task_test.go
View File

@@ -1,15 +1,11 @@
package tasksch
import (
"errors"
"fmt"
"testing"
"git.rosy.net.cn/baseapi/utils"
)
func TestTaskError(t *testing.T) {
err := NewTaskError("test", errors.New("hello"))
fmt.Println(utils.Format4Output(err, false))
fmt.Println(err.Error())
// err := NewTaskError("test", errors.New("hello"))
// fmt.Println(utils.Format4Output(err, false))
// fmt.Println(err.Error())
}