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
2020-01-17 22:16:00 +08:00
parent 4f30241f2a
commit 1a2f653561
3 changed files with 32 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -186,6 +186,5 @@ func (task *ParallelTask) Run() {
} }
func (t *ParallelTask) AddChild(task ITask) ITask { func (t *ParallelTask) AddChild(task ITask) ITask {
task.SetParent(t)
return t.BaseTask.AddChild(task) return t.BaseTask.AddChild(task)
} }

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

@@ -69,6 +69,5 @@ func (task *SeqTask) Run() {
} }
func (t *SeqTask) AddChild(task ITask) ITask { func (t *SeqTask) AddChild(task ITask) ITask {
task.SetParent(t)
return t.BaseTask.AddChild(task) return t.BaseTask.AddChild(task)
} }

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

@@ -63,6 +63,7 @@ type ITask interface {
// GetDetailErrList() []error // GetDetailErrList() []error
GetLeafResult() (finishedItemCount, failedItemCount int) GetLeafResult() (finishedItemCount, failedItemCount int)
AddBatchErr(err error) AddBatchErr(err error)
AddFailedList(failedList ...interface{}) AddFailedList(failedList ...interface{})
GetFailedList() (failedList []interface{}) GetFailedList() (failedList []interface{})
SetFinishHook(func(task ITask)) SetFinishHook(func(task ITask))
@@ -247,6 +248,8 @@ func (t *BaseTask) GetStatus() int {
} }
func (t *BaseTask) AddChild(task ITask) ITask { func (t *BaseTask) AddChild(task ITask) ITask {
task.SetParent(t)
t.locker.Lock() t.locker.Lock()
defer t.locker.Unlock() defer t.locker.Unlock()
@@ -368,22 +371,41 @@ func (t *BaseTask) SetFinishHook(hook func(task ITask)) {
t.finishHook = hook t.finishHook = hook
} }
// 此函数非线程安全，只能在确定任务结束后调用
func (t *BaseTask) GetFailedList() (failedList []interface{}) { func (t *BaseTask) GetFailedList() (failedList []interface{}) {
t.locker.RLock() // t.locker.RLock()
failedList = append(failedList, t.FailedList...) // failedList = append(failedList, t.FailedList...)
t.locker.RUnlock() // t.locker.RUnlock()
for _, v := range t.Children { // for _, v := range t.Children {
failedList = append(failedList, v.GetFailedList()...) // failedList = append(failedList, v.GetFailedList()...)
} // }
return failedList // return failedList
// t.locker.RLock()
// defer t.locker.RUnlock()
// if len(t.FailedList) > 0 {
// failedList = make([]interface{}, len(t.FailedList))
// copy(failedList, t.FailedList)
// }
// return failedList
return t.FailedList
} }
func (t *BaseTask) AddFailedList(failedList ...interface{}) { func (t *BaseTask) AddFailedList(failedList ...interface{}) {
if len(failedList) > 0 { if len(failedList) > 0 {
t.locker.Lock() // t.locker.Lock()
defer t.locker.Unlock() // defer t.locker.Unlock()
t.FailedList = append(t.FailedList, failedList...) // t.FailedList = append(t.FailedList, failedList...)
if t.parent == nil || t.finishHook != nil {
t.locker.Lock()
defer t.locker.Unlock()
t.FailedList = append(t.FailedList, failedList...)
} else {
t.parent.AddFailedList(failedList...)
}
} }
} }