Merge pull request #2 from memory-overflow/dev/demo-video-cut

Dev/demo video cut

Author: memory-overflow

container/memory_container/queue_container.go

@@ -137,7 +137,7 @@ func (q *queueContainer) ToFailedStatus(ctx context.Context, task *lighttasksche
 	}
 	log.Printf("failed task %s, reason %v", task.TaskId, reason)
 	task.TaskStatus = lighttaskscheduler.TASK_STATUS_FAILED
-	return
+	return task, nil
 }
 
 // ToExportStatus 转移到数据导出状态

example/actuator/example_actuator.go

@@ -0,0 +1,117 @@
+package addexample
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"math/rand"
+	"sync"
+	"time"
+
+	framework "github.com/memory-overflow/light-task-scheduler"
+)
+
+// AddActuator 加法执行器，同步任务异步化的示例
+type AddActuator struct {
+	runningTask sync.Map // taskId -> framework.AsyncTaskStatus 映射
+	resultMap   sync.Map // taskId -> int32 映射
+}
+
+// Init 任务在被调度前的初始化工作
+func (a *AddActuator) Init(ctx context.Context, task *framework.Task) (
+	newTask *framework.Task, err error) {
+	// 无初始化工作
+	return task, nil
+}
+
+func (add *AddActuator) work(taskId string, a, b int32) {
+	time.Sleep(time.Duration(rand.Intn(4000))*time.Millisecond + 2*time.Second) // 25% 概率超时
+	if _, ok := add.runningTask.Load(taskId); !ok {
+		// 任务可能因为超时被删除，不处理
+		return
+	}
+	newStatus := framework.AsyncTaskStatus{
+		TaskStatus: framework.TASK_STATUS_SUCCESS,
+		Progress:   100.0,
+	}
+	add.resultMap.Store(taskId, a+b)
+	add.runningTask.Store(taskId, newStatus)
+}
+
+// Start 执行任务
+func (a *AddActuator) Start(ctx context.Context, ftask *framework.Task) (
+	newTask *framework.Task, ignoreErr bool, err error) {
+	if _, ok := a.runningTask.Load(ftask.TaskId); ok {
+		// 任务已经在执行中，不能重复执行
+		return ftask, false, fmt.Errorf("task %s is running", ftask.TaskId)
+	}
+	task, ok := ftask.TaskItem.(AddTask)
+	if !ok {
+		return ftask, false, fmt.Errorf("TaskItem not be set to AddTask")
+	}
+	ftask.TaskStatus = framework.TASK_STATUS_RUNNING
+	ftask.TaskStartTime = time.Now()
+	a.runningTask.Store(ftask.TaskId, framework.AsyncTaskStatus{
+		TaskStatus: framework.TASK_STATUS_RUNNING,
+		Progress:   0.0,
+	})
+	go a.work(ftask.TaskId, task.A, task.B)
+	log.Printf("start success, taskid: %s\n", ftask.TaskId)
+	return ftask, false, nil
+}
+
+// ExportOutput 导出任务输出，自行处理任务结果
+func (a *AddActuator) ExportOutput(ctx context.Context, ftask *framework.Task) error {
+	task, ok := ftask.TaskItem.(AddTask)
+	if !ok {
+		return fmt.Errorf("TaskItem not be set to AddTask")
+	}
+	res, ok := a.resultMap.Load(ftask.TaskId)
+	if !ok {
+		return fmt.Errorf("not found result for task %s", ftask.TaskId)
+	}
+	log.Printf("finished task %s: %d + %d = %d \n", ftask.TaskId, task.A, task.B, res)
+	a.resultMap.Delete(ftask.TaskId) // delete result after export
+	return nil
+}
+
+// Stop 停止任务
+func (a *AddActuator) Stop(ctx context.Context, ftask *framework.Task) error {
+	// 同步任务无法真正暂停，只能删除状态
+	a.runningTask.Delete(ftask.TaskId)
+	a.resultMap.Delete(ftask.TaskId)
+	return nil
+}
+
+// GetAsyncTaskStatus 批量获取任务状态
+func (a *AddActuator) GetAsyncTaskStatus(ctx context.Context, ftasks []framework.Task) (
+	status []framework.AsyncTaskStatus, err error) {
+	for _, ftask := range ftasks {
+		fstatus, ok := a.runningTask.Load(ftask.TaskId)
+		if !ok {
+			status = append(status, framework.AsyncTaskStatus{
+				TaskStatus:   framework.TASK_STATUS_FAILED,
+				FailedReason: errors.New("同步任务未找到"),
+				Progress:     float32(0.0),
+			})
+		} else {
+			if fstatus.(framework.AsyncTaskStatus).TaskStatus != framework.TASK_STATUS_RUNNING {
+				a.runningTask.Delete(ftask.TaskId) // delete task status after query if task finished
+			}
+			status = append(status, fstatus.(framework.AsyncTaskStatus))
+		}
+	}
+	return status, nil
+}
+
+// GetOutput ...
+func (e *AddActuator) GetOutput(ctx context.Context, task *framework.Task) (
+	data interface{}, err error) {
+	return nil, nil
+}
+
+// GetOutput ...
+func (e *AddActuator) Delete(ctx context.Context, task *framework.Task) (err error) {
+	return nil
+}

example/add_example/add/add_actuator.go

@@ -0,0 +1,9 @@
+package addexample
+
+import "time"
+
+// AddTask add 任务结构
+type AddTask struct {
+	StartTime time.Time
+	A, B      int32
+}

example/add_example/add/add_task.go

@@ -10,36 +10,36 @@ import (
 
 	lighttaskscheduler "github.com/memory-overflow/light-task-scheduler"
 	memeorycontainer "github.com/memory-overflow/light-task-scheduler/container/memory_container"
-	"github.com/memory-overflow/light-task-scheduler/example/actuator"
-	service "github.com/memory-overflow/light-task-scheduler/example/add_service"
-	"github.com/memory-overflow/light-task-scheduler/example/task"
+
+	addexample "github.com/memory-overflow/light-task-scheduler/example/add_example/add"
 )
 
 func main() {
-	go service.StartServer()
 	container := memeorycontainer.MakeQueueContainer(10000, 100*time.Millisecond)
-	actuator := actuator.MakeExampleActuator()
+	actuator := &addexample.AddActuator{}
 	sch := lighttaskscheduler.MakeNewScheduler(
 		context.Background(),
 		container, actuator,
 		lighttaskscheduler.Config{
 			TaskLimit:    5,
-			ScanInterval: 50 * time.Millisecond})
+			ScanInterval: 50 * time.Millisecond,
+			TaskTimeout:  5 * time.Second, // 5s 超时
+		},
+	)
 
 	var c chan os.Signal
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
-	for i := 0; i < 1000; i++ {
+	for i := 0; i < 10; i++ {
 		select {
 		case <-c:
 			return
 		default:
 			sch.AddTask(context.Background(),
 				lighttaskscheduler.Task{
 					TaskId: strconv.Itoa(i),
-					TaskItem: task.ExampleTask{
-						TaskId: uint32(i),
-						A:      r.Int31() % 1000,
-						B:      r.Int31() % 1000,
+					TaskItem: addexample.AddTask{
+						A: r.Int31() % 1000,
+						B: r.Int31() % 1000,
 					},
 				})
 		}