解决冲突，合并提交

business/jxutils/jxutils.go

@@ -9,7 +9,6 @@ import (
 	"regexp"
 	"sort"
 	"strings"
-	"sync"
 	"time"
 	"io/ioutil"
 
@@ -28,11 +27,6 @@ var (
 	skuNamePat  *regexp.Regexp
 )
 
-type SyncMapWithTimeout struct {
-	sync.Map
-	timers sync.Map
-}
-
 type OrderSkuList []*model.OrderSku
 
 func (l OrderSkuList) Len() int {
@@ -61,22 +55,6 @@ func init() {
 	skuNamePat = regexp.MustCompile(`([\(（\[【][^\(（\[【\)）\]】]*[\)）\]】])?(.*?)([(（].*[)）])?\s*约?([1-9][\d\.]*)(g|G|kg|kG|Kg|KG|l|L|ml|mL|Ml|ML|克)\s*([(（].*[)）])?\s*(?:\/|／|)\s*([^\s（）()]{0,2})\s*([（(].*[）)])?$`)
 }
 
-func (m *SyncMapWithTimeout) StoreWithTimeout(key, value interface{}, timeout time.Duration) {
-	m.Map.Store(key, value)
-	m.timers.Store(key, utils.AfterFuncWithRecover(timeout, func() {
-		m.Delete(key)
-	}))
-}
-
-func (m *SyncMapWithTimeout) Delete(key interface{}) {
-	m.Map.Delete(key)
-	if value, ok := m.timers.Load(key); ok {
-		timer := value.(*time.Timer)
-		timer.Stop()
-	}
-	m.timers.Delete(key)
-}
-
 func getJxStoreIDFromOrder(order *model.GoodsOrder) (retVal int) {
 	if order.JxStoreID != 0 {
 		return order.JxStoreID

business/jxutils/jxutils_sync_map.go

@@ -0,0 +1,29 @@
+package jxutils
+
+import (
+	"sync"
+	"time"
+
+	"git.rosy.net.cn/baseapi/utils"
+)
+
+type SyncMapWithTimeout struct {
+	sync.Map
+	timers sync.Map
+}
+
+func (m *SyncMapWithTimeout) StoreWithTimeout(key, value interface{}, timeout time.Duration) {
+	m.Map.Store(key, value)
+	m.timers.Store(key, utils.AfterFuncWithRecover(timeout, func() {
+		m.Delete(key)
+	}))
+}
+
+func (m *SyncMapWithTimeout) Delete(key interface{}) {
+	m.Map.Delete(key)
+	if value, ok := m.timers.Load(key); ok {
+		timer := value.(*time.Timer)
+		timer.Stop()
+	}
+	m.timers.Delete(key)
+}

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

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

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

business/jxutils/tasksch/task_man.go

@@ -2,45 +2,44 @@ package tasksch
 
 import (
 	"sort"
-	"sync"
 	"time"
+
+	"git.rosy.net.cn/jx-callback/business/jxutils"
 )
 
 var (
 	defTaskMan   TaskMan
 	defLastHours = 24
+	maxStoreTime = 48 * time.Hour // 最多存两天时间
 )
 
 type TaskMan struct {
-	taskList map[string]ITask
-	locker   sync.RWMutex
+	taskMap jxutils.SyncMapWithTimeout
 }
 
 func init() {
-	defTaskMan.taskList = make(map[string]ITask)
 }
 
 func (m *TaskMan) GetTasks(taskID string, fromStatus, toStatus int, lastHours int, createdBy string) (taskList TaskList) {
-	m.locker.RLock()
-	defer m.locker.RUnlock()
 	if lastHours == 0 {
 		lastHours = defLastHours
 	}
 	lastTime := time.Now().Add(time.Duration(-lastHours) * time.Hour).Unix()
-	for k, v := range m.taskList {
+	m.taskMap.Range(func(key, value interface{}) bool {
+		k := key.(string)
+		v := value.(ITask)
 		status := v.GetStatus()
 		if !((createdBy != "" && createdBy != v.GetCreatedBy()) || (taskID != "" && taskID != k) || status < fromStatus || status > toStatus || v.GetCreatedAt().Unix() < lastTime) {
 			taskList = append(taskList, v)
 		}
-	}
+		return true
+	})
 	sort.Sort(TaskList(taskList))
 	return taskList
 }
 
 func (m *TaskMan) ManageTask(task ITask) ITask {
-	m.locker.Lock()
-	defer m.locker.Unlock()
-	m.taskList[task.GetID()] = task
+	m.taskMap.StoreWithTimeout(task.GetID(), task, maxStoreTime)
 	return task
 }
 
@@ -53,7 +52,7 @@ func ManageTask(task ITask) ITask {
 }
 
 func IsTaskRunning(taskID string) bool {
-	if taskList := GetTasks(taskID, TaskStatusBegin, TaskStatusWorking, 36, ""); len(taskList) > 0 {
+	if taskList := GetTasks(taskID, TaskStatusBegin, TaskStatusWorking, 0, ""); len(taskList) > 0 {
 		return true
 	}
 	return false