增加任务依赖

README.md

@@ -10,6 +10,7 @@
 * 任务执行失败重试设置
 * 任务超时设置
 * 延时任务
+* 任务依赖
 * 任务类型
     * shell任务
     > 在远程服务器上执行shell命令
@@ -65,6 +66,7 @@
 ### 命令
 
 * gocron web
+    * --host 默认0.0.0.0
     * -p 端口, 指定端口, 默认5920
     * -e 指定运行环境, dev|test|prod, dev模式下可查看更多日志信息, 默认prod
     * -d 后台运行

cmd/web.go

@@ -31,6 +31,11 @@ var CmdWeb = cli.Command{
     Usage:  "run web server",
     Action: runWeb,
     Flags: []cli.Flag{
+        cli.StringFlag{
+            Name: "host",
+            Value: "0.0.0.0",
+            Usage: "bind host",
+        },
         cli.IntFlag{
             Name:  "port,p",
             Value: DefaultPort,
@@ -66,9 +71,10 @@ func runWeb(ctx *cli.Context) {
     routers.Register(m)
     // 注册中间件.
     routers.RegisterMiddleware(m)
+    host := parseHost(ctx)
     port := parsePort(ctx)
     fmt.Println("server start")
-    m.Run(port)
+    m.Run(host, port)
 }
 
 func becomeDaemon(ctx *cli.Context) {
@@ -151,6 +157,14 @@ func parsePort(ctx *cli.Context) int {
     return port
 }
 
+func parseHost(ctx *cli.Context) string  {
+    if ctx.IsSet("host") {
+        return ctx.String("host")
+    }
+
+    return "0.0.0.0"
+}
+
 func setEnvironment(ctx *cli.Context)  {
     var env string = "prod"
     if ctx.IsSet("env") {

models/model.go

@@ -10,6 +10,7 @@ import (
     "github.com/ouqiang/gocron/modules/logger"
     "github.com/ouqiang/gocron/modules/app"
     "strconv"
+    "time"
 )
 
 type Status int8
@@ -93,6 +94,7 @@ func CreateDb() *xorm.Engine {
         engine.Logger().SetLevel(core.LOG_DEBUG)
     }
 
+    go keepDbAlived(engine)
 
     return engine
 }
@@ -139,3 +141,11 @@ func getDbConfig() map[string]string {
 
     return db
 }
+
+func keepDbAlived(engine *xorm.Engine)  {
+    t := time.Tick(180 * time.Second)
+    for {
+        <- t
+        engine.Ping()
+    }
+}

models/task.go

@@ -4,6 +4,7 @@ import (
     "time"
     "github.com/go-xorm/xorm"
     "errors"
+    "strings"
 )
 
 type TaskProtocol int8
@@ -13,22 +14,39 @@ const (
     TaskRPC  // RPC方式执行命令
 )
 
+type TaskLevel int8
+
+const (
+    TaskLevelParent TaskLevel = 1 // 父任务
+    TaskLevelChild  TaskLevel = 2 // 子任务(依赖任务)
+)
+
+type TaskDependencyStatus int8
+
+const (
+    TaskDependencyStatusStrong TaskDependencyStatus = 1 // 强依赖
+    TaskDependencyStatusWeak   TaskDependencyStatus = 2 // 弱依赖
+)
+
 // 任务
 type Task struct {
     Id       int       `xorm:"int pk autoincr"`
     Name     string    `xorm:"varchar(32) notnull"`              // 任务名称
+    Level    TaskLevel     `xorm:"smallint notnull index default 1"`     // 任务等级 1: 主任务 2: 依赖任务
+    DependencyTaskId string `xorm:"varchar(64) notnull default ''"` // 依赖任务ID,多个ID逗号分隔
+    DependencyStatus TaskDependencyStatus  `xorm:"smallint notnull default 1"`   // 依赖关系 1:强依赖 主任务执行成功, 依赖任务才会被执行 2:弱依赖
     Spec     string    `xorm:"varchar(64) notnull"`              // crontab
-    Protocol TaskProtocol  `xorm:"tinyint notnull"`              // 协议 1:http 2:系统命令
+    Protocol TaskProtocol  `xorm:"tinyint notnull index"`              // 协议 1:http 2:系统命令
     Command  string    `xorm:"varchar(256) notnull"`             // URL地址或shell命令
     Timeout  int       `xorm:"mediumint notnull default 0"`      // 任务执行超时时间(单位秒),0不限制
     Multi    int8      `xorm:"tinyint notnull default 1"`        // 是否允许多实例运行
     RetryTimes int8    `xorm:"tinyint notnull default 0"`         // 重试次数
-    HostId   int16    `xorm:"smallint notnull default 0"`        // RPC host id，
+    HostId   int16    `xorm:"smallint notnull index default 0"`        // RPC host id，
     NotifyStatus int8  `xorm:"smallint notnull default 1"`       // 任务执行结束是否通知 0: 不通知 1: 失败通知 2: 执行结束通知
     NotifyType int8 `xorm:"smallint notnull default 0"`  // 通知类型 1: 邮件 2: slack
     NotifyReceiverId string `xorm:"varchar(256) notnull default '' "` // 通知接受者ID, setting表主键ID，多个ID逗号分隔
     Remark   string    `xorm:"varchar(100) notnull default ''"`  // 备注
-    Status   Status    `xorm:"tinyint notnull default 0"`        // 状态 1:正常 0:停止
+    Status   Status    `xorm:"tinyint notnull index default 0"`        // 状态 1:正常 0:停止
     Created  time.Time `xorm:"datetime notnull created"`         // 创建时间
     Deleted  time.Time `xorm:"datetime deleted"`                 // 删除时间
     BaseModel `xorm:"-"`
@@ -59,6 +77,7 @@ func (task *Task) Create() (insertId int, err error) {
 func (task *Task) CreateTestTask() {
     // HTTP任务
     task.Name = "测试HTTP任务"
+    task.Level = TaskLevelParent
     task.Protocol = TaskHTTP
     task.Spec = "*/30 * * * * *"
     // 查询IP地址区域信息
@@ -68,7 +87,9 @@ func (task *Task) CreateTestTask() {
 }
 
 func (task *Task) UpdateBean(id int) (int64, error)  {
-    return Db.ID(id).Cols("name,spec,protocol,command,timeout,multi,retry_times,host_id,remark,notify_status,notify_type,notify_receiver_id").Update(task)
+    return Db.ID(id).
+    Cols("name,spec,protocol,command,timeout,multi,retry_times,host_id,remark,notify_status,notify_type,notify_receiver_id, dependency_task_id, dependency_status").
+    Update(task)
 }
 
 // 更新
@@ -95,7 +116,11 @@ func (task *Task) Enable(id int) (int64, error) {
 func (task *Task) ActiveList() ([]TaskHost, error) {
     list := make([]TaskHost, 0)
     fields := "t.*, host.alias,host.name,host.port"
-    err := Db.Alias("t").Join("LEFT", hostTableName(), "t.host_id=host.id").Where("t.status = ?", Enabled).Cols(fields).Find(&list)
+    err := Db.Alias("t").
+            Join("LEFT", hostTableName(), "t.host_id=host.id").
+            Where("t.status = ? AND t.level = ?", Enabled, TaskLevelParent).
+            Cols(fields).
+            Find(&list)
 
     return list, err
 }
@@ -104,7 +129,11 @@ func (task *Task) ActiveList() ([]TaskHost, error) {
 func (task *Task) ActiveListByHostId(hostId int16) ([]TaskHost, error) {
     list := make([]TaskHost, 0)
     fields := "t.*, host.alias,host.name,host.port"
-    err := Db.Alias("t").Join("LEFT", hostTableName(), "t.host_id=host.id").Where("t.status = ? AND t.host_id = ?", Enabled, hostId).Cols(fields).Find(&list)
+    err := Db.Alias("t").
+        Join("LEFT", hostTableName(), "t.host_id=host.id").
+        Where("t.status = ? AND t.host_id = ? AND t.level = ?", Enabled, hostId, TaskLevelParent).
+        Cols(fields).
+        Find(&list)
 
     return list, err
 }
@@ -158,6 +187,28 @@ func (task *Task) List(params CommonMap) ([]TaskHost, error) {
     return list, err
 }
 
+// 获取依赖任务列表
+func (task *Task) GetDependencyTaskList(ids string) ([]TaskHost, error) {
+    list := make([]TaskHost, 0)
+    if ids == "" {
+        return list, nil
+    }
+    idList := strings.Split(ids, ",")
+    taskIds := make([]interface{}, len(idList))
+    for i,  v := range idList {
+        taskIds[i] = v
+    }
+    fields := "t.*, host.alias,host.name,host.port"
+    err := Db.Alias("t").
+            Join("LEFT", hostTableName(), "t.host_id=host.id").
+            Where("t.level = ?", TaskLevelChild).
+            In("t.id", taskIds).
+            Cols(fields).
+            Find(&list)
+
+    return list, err
+}
+
 func (task *Task) Total(params CommonMap) (int64, error) {
     session := Db.Alias("t").Join("LEFT", hostTableName(), "t.host_id=host.id")
     task.parseWhere(session, params)

routers/task/task.go

@@ -18,8 +18,11 @@ import (
 
 type TaskForm struct {
     Id int
+    Level models.TaskLevel `binding:"Required;In(1,2)"`
+    DependencyStatus models.TaskDependencyStatus
+    DependencyTaskId string
     Name string `binding:"Required;MaxSize(32)"`
-    Spec string `binding:"Required;MaxSize(64)"`
+    Spec string
     Protocol models.TaskProtocol `binding:"In(1,2)"`
     Command string `binding:"Required;MaxSize(256)"`
     Timeout int `binding:"Range(0,86400)"`
@@ -99,10 +102,6 @@ func Store(ctx *macaron.Context, form TaskForm) string  {
     json := utils.JsonResponse{}
     taskModel := models.Task{}
     var id int = form.Id
-    _, err := cron.Parse(form.Spec)
-    if err != nil {
-        return json.CommonFailure("crontab表达式解析失败", err)
-    }
     nameExists, err := taskModel.NameExist(form.Name, form.Id)
     if err != nil {
         return json.CommonFailure(utils.FailureContent, err)
@@ -134,8 +133,11 @@ func Store(ctx *macaron.Context, form TaskForm) string  {
     taskModel.NotifyType = form.NotifyType - 1
     taskModel.NotifyReceiverId = form.NotifyReceiverId
     taskModel.Spec = form.Spec
+    taskModel.Level = form.Level
+    taskModel.DependencyStatus = form.DependencyStatus
+    taskModel.DependencyTaskId = strings.TrimSpace(form.DependencyTaskId)
     if taskModel.NotifyStatus > 0 && taskModel.NotifyReceiverId == "" {
-        return json.CommonFailure("请至少选择一个接收者")
+        return json.CommonFailure("至少选择一个通知接收者")
     }
     if taskModel.Protocol == models.TaskHTTP {
         command := strings.ToLower(taskModel.Command)
@@ -151,6 +153,27 @@ func Store(ctx *macaron.Context, form TaskForm) string  {
         return json.CommonFailure("任务重试次数取值0-10")
     }
 
+    if (taskModel.DependencyStatus != models.TaskDependencyStatusStrong &&
+        taskModel.DependencyStatus != models.TaskDependencyStatusWeak) {
+        return json.CommonFailure("请选择依赖关系")
+    }
+
+    if taskModel.Level == models.TaskLevelParent {
+        _, err = cron.Parse(form.Spec)
+        if err != nil {
+            return json.CommonFailure("crontab表达式解析失败", err)
+        }
+    } else {
+        taskModel.DependencyTaskId = ""
+        taskModel.Spec = ""
+    }
+
+    if id > 0 && taskModel.DependencyTaskId != "" {
+        dependencyTaskIds := strings.Split(taskModel.DependencyTaskId, ",")
+        if utils.InStringSlice(dependencyTaskIds, strconv.Itoa(id)) {
+            return json.CommonFailure("不允许设置当前任务为子任务")
+        }
+    }
 
     if id == 0 {
         // 任务添加后开始调度执行
@@ -165,7 +188,7 @@ func Store(ctx *macaron.Context, form TaskForm) string  {
     }
 
     status, err := taskModel.GetStatus(id)
-    if status == models.Enabled {
+    if status == models.Enabled && taskModel.Level == models.TaskLevelParent {
         addTaskToTimer(id)
     }
 

service/task.go

@@ -13,6 +13,7 @@ import (
     "sync"
     rpcClient "github.com/ouqiang/gocron/modules/rpc/client"
     pb "github.com/ouqiang/gocron/modules/rpc/proto"
+    "strings"
 )
 
 // 定时任务调度管理器
@@ -47,13 +48,16 @@ func (c *TaskCount) Num() int  {
     return c.num
 }
 
-// 任务ID作为Key, 不会出现并发写, 不加锁
+// 任务ID作为Key
 type Instance struct {
     Status map[int]bool
+    sync.RWMutex
 }
 
 // 是否有任务处于运行中
 func (i *Instance) has(key int) bool {
+    i.RLock()
+    defer i.RUnlock()
     running, ok := i.Status[key]
     if ok && running {
         return true
@@ -63,11 +67,15 @@ func (i *Instance) has(key int) bool {
 }
 
 func (i *Instance) add(key int)  {
+    i.Lock()
+    defer i.Unlock()
     i.Status[key] = true
 }
 
 func (i *Instance) done(key int)  {
-    i.Status[key] = false
+    i.Lock()
+    defer i.Unlock()
+    delete(i.Status, key)
 }
 
 type Task struct{}
@@ -82,7 +90,7 @@ type TaskResult struct {
 func (task *Task) Initialize() {
     Cron = cron.New()
     Cron.Start()
-    runInstance = Instance{make(map[int]bool)}
+    runInstance = Instance{make(map[int]bool), sync.RWMutex{}}
     TaskNum = TaskCount{0, sync.RWMutex{}}
 
     taskModel := new(models.Task)
@@ -107,6 +115,10 @@ func (task *Task) BatchAdd(tasks []models.TaskHost)  {
 
 // 添加任务
 func (task *Task) Add(taskModel models.TaskHost) {
+    if taskModel.Level == models.TaskLevelChild {
+        logger.Errorf("添加任务失败#不允许添加子任务到调度器#任务Id-%d", taskModel.Id);
+        return
+    }
     taskFunc := createJob(taskModel)
     if taskFunc == nil {
         logger.Error("创建任务处理Job失败,不支持的任务协议#", taskModel.Protocol)
@@ -239,6 +251,7 @@ func createHandler(taskModel models.TaskHost) Handler  {
     return handler;
 }
 
+// 任务前置操作
 func beforeExecJob(taskModel models.TaskHost) (taskLogId int64)  {
     if taskModel.Multi == 0 && runInstance.has(taskModel.Id) {
         createTaskLog(taskModel, models.Cancel)
@@ -258,6 +271,7 @@ func beforeExecJob(taskModel models.TaskHost) (taskLogId int64)  {
     return taskLogId
 }
 
+// 任务执行后置操作
 func afterExecJob(taskModel models.TaskHost, taskResult TaskResult, taskLogId int64)  {
     if taskResult.Err != nil {
         taskResult.Result = taskResult.Err.Error() + "\n" + taskResult.Result
@@ -267,7 +281,47 @@ func afterExecJob(taskModel models.TaskHost, taskResult TaskResult, taskLogId in
         logger.Error("任务结束#更新任务日志失败-", err)
     }
 
-    SendNotification(taskModel, taskResult)
+    // 发送邮件
+    go SendNotification(taskModel, taskResult)
+    // 执行依赖任务
+    go execDependencyTask(taskModel, taskResult)
+}
+
+// 执行依赖任务, 多个任务并发执行
+func execDependencyTask(taskModel models.TaskHost, taskResult TaskResult)  {
+    // 父任务才能执行子任务
+    if taskModel.Level != models.TaskLevelParent {
+        return
+    }
+
+    // 是否存在子任务
+    dependencyTaskId := strings.TrimSpace(taskModel.DependencyTaskId)
+    if dependencyTaskId == "" {
+        return
+    }
+
+    // 父子任务关系为强依赖, 父任务执行失败, 不执行依赖任务
+    if taskModel.DependencyStatus == models.TaskDependencyStatusStrong && taskResult.Err != nil {
+        logger.Infof("父子任务为强依赖关系, 父任务执行失败, 不运行依赖任务#主任务ID-%d", taskModel.Id)
+        return
+    }
+
+    // 获取子任务
+    model := new(models.Task)
+    tasks , err := model.GetDependencyTaskList(dependencyTaskId)
+    if err != nil {
+        logger.Errorf("获取依赖任务失败#主任务ID-%d#%s", taskModel.Id, err.Error())
+        return
+    }
+    if len(tasks) == 0 {
+        logger.Errorf("依赖任务列表为空#主任务ID-%d", taskModel.Id)
+    }
+
+    serviceTask := new(Task)
+    for _, task := range tasks {
+        task.Spec = fmt.Sprintf("依赖任务(主任务ID-%d)", taskModel.Id)
+        serviceTask.Run(task)
+    }
 }
 
 // 发送任务结果通知

templates/task/index.html

@@ -54,6 +54,7 @@
             <tr>
                 <th>任务ID</th>
                 <th>任务名称</th>
+                <th>任务类型</th>
                 <th>cron表达式</th>
                 <th>执行方式</th>
                 <th>超时时间</th>
@@ -69,21 +70,28 @@
                     <tr>
                         <td>{{{.Id}}}</td>
                         <td>{{{.Task.Name}}}</td>
+                        <td>{{{if eq .Level 1}}}主任务{{{else}}}子任务{{{end}}}</td>
                         <td>{{{.Spec}}}</td>
                         <td>{{{if eq .Protocol 1}}} HTTP {{{else if eq .Protocol 2}}} SHELL {{{end}}}</td>
                         <td>{{{if eq .Timeout -1}}}后台运行{{{else if gt .Timeout 0}}}{{{.Timeout}}}秒{{{else}}}不限制{{{end}}}</td>
                         <td>{{{.RetryTimes}}}</td>
                         <td>{{{if gt .Multi 0}}}否{{{else}}}是{{{end}}}</td>
                         <td>{{{.Alias}}}-{{{.Name}}}</td>
-                        <td>{{{if eq .Status 1}}}<span style="color: green;">激活</span>{{{else}}}<span style="color: red;">停止<span>{{{end}}}</td>
+                        <td>
+                            {{{if eq .Level 1}}}
+                                {{{if eq .Status 1}}}<span style="color: green;">激活</span>{{{else}}}<span style="color: red;">停止<span>{{{end}}}
+                            {{{end}}}
+                        </td>
                         <td>
                             <div class="ui buttons operation">
                                 <a class="ui purple  button" href="/task/edit/{{{.Id}}}">编辑</a>
+                                {{{if eq .Level 1}}}
                                     {{{if eq .Status 1}}}
                                     <button class="ui primary button" @click="changeStatus({{{.Id}}},{{{.Status}}})">停止</button>
                                     {{{else}}}
                                     <button class="ui blue button" @click="changeStatus({{{.Id}}},{{{.Status}}})">激活 </button>
                                     {{{end}}}
+                                {{{end}}}
                                 <button class="ui positive button" @click="remove({{{.Id}}})">删除</button> <br>
                                 <button class="ui twitter button" @click="run({{{.Id}}})">手动运行</button>
                                 <a class="ui instagram button" href="/task/log?task_id={{{.Id}}}">查看日志</a>
@@ -97,7 +105,6 @@
     </div>
 </div>
 
-
 <script type="text/javascript">
     $('.ui.checkbox').checkbox();
 

templates/task/task_form.html

@@ -19,7 +19,52 @@
                     <div class="content">任务名称</div>
                 </label>
                 <div class="ui small input">
-                    <input type="text" name="name" value="{{{.Task.Task.Name}}}">
+                    <input type="text" name="name" placeholder="订单量统计" value="{{{.Task.Task.Name}}}">
+                </div>
+            </div>
+        </div>
+        <div class="two fields">
+            <div class="field">
+                <label>
+                    <div class="content">任务类型</div>
+                    <div class="ui message">
+                        主任务可以配置多个子任务, 当主任务执行完成后，自动执行子任务<br>
+                        任务类型新增后不能变更
+                    </div>
+                </label>
+                <select name="level" id="level" {{{if .Task}}}disabled="disabled"{{{end}}}>
+                    <option value="1" {{{if .Task}}} {{{if eq .Task.Level 1}}}selected{{{end}}} {{{end}}}>主任务</option>
+                    <option value="2" {{{if .Task}}} {{{if eq .Task.Level 2}}}selected{{{end}}} {{{end}}}>子任务</option>
+                </select>
+            </div>
+        </div>
+        <div id="parent-task">
+            <div class="two fields">
+                <div class="field">
+                    <label>
+                        <div class="content">依赖关系</div>
+                        <div class="ui message">
+                            强依赖: 主任务执行成功，才会运行子任务 <br>
+                            弱依赖: 无论主任务执行是否成功，都会运行子任务
+                        </div>
+                    </label>
+                    <select name="dependency_status" id="dependency_status">
+                        <option value="1" {{{if .Task}}} {{{if eq .Task.DependencyStatus 1}}}selected{{{end}}} {{{end}}}>强依赖</option>
+                        <option value="2" {{{if .Task}}} {{{if eq .Task.DependencyStatus 2}}}selected{{{end}}} {{{end}}}>弱依赖</option>
+                    </select>
+                </div>
+            </div>
+            <div class="two fields">
+                <div class="field">
+                    <label>
+                        <div class="content">子任务ID</div>
+                        <div class="ui message">
+                            多个任务ID逗号分隔 <br>
+                            子任务并发执行
+                        </div>
+                    </label>
+                    <div class="ui small input">
+                        <input type="text" name="dependency_task_id" placeholder="可选" value="{{{.Task.DependencyTaskId}}}">
                     </div>
                 </div>
             </div>
@@ -35,6 +80,7 @@
                     </div>
                 </div>
             </div>
+        </div>
         <div class="three fields">
             <div class="field">
                 <label>执行方式</label>
@@ -46,7 +92,7 @@
         </div>
         <div class="three fields" id="hostField">
             <div class="field">
-                <label>主机</label>
+                <label>任务执行器</label>
                 <select name="host_id" id="hostId">
                     <option value="">选择主机</option>
                     {{{range $i, $v := .Hosts}}}
@@ -64,13 +110,13 @@
         </div>
         <div class="six fields">
             <div class="field">
-                <label>任务超时时间(秒)</label>
+                <label>任务超时时间(秒, 0-86400)</label>
                 <input type="text"  name="timeout" placeholder="默认0, 不限制" value="{{{if .Task}}} {{{.Task.Timeout}}} {{{else}}} 0 {{{end}}}">
             </div>
         </div>
         <div class="six fields">
             <div class="field">
-                <label>任务失败重试次数</label>
+                <label>任务失败重试次数 (0-10)</label>
                 <input type="text"  name="retry_times" placeholder="默认0, 不重试" value="{{{if .Task}}} {{{.Task.RetryTimes}}} {{{else}}} 0 {{{end}}}">
             </div>
         </div>
@@ -107,7 +153,7 @@
         <div class="two fields">
             <div class="field">
                 <label>备注</label>
-                <textarea rows="5" name="remark">{{{.Task.Remark}}}</textarea>
+                <textarea rows="5" name="remark" placeholder="统计昨天的订单量">{{{.Task.Remark}}}</textarea>
             </div>
         </div>
         <div class="ui primary submit button">保存</div>   <a class="ui button" onclick="location.href='/task';">取消</a>
@@ -144,6 +190,7 @@
 <script type="text/javascript">
     $(function() {
         changeCommandPlaceholder();
+        changeLevel();
         changeProtocol();
         showNotify();
     });
@@ -153,7 +200,9 @@
         changeProtocol();
     });
 
-
+    $('#level').change(function() {
+        changeLevel();
+    });
 
     $('#task-status').change(function() {
         var selected = $(this).val();
@@ -269,6 +318,19 @@
         return receivers.join(",");
     }
 
+    function changeLevel() {
+        var selected = $('#level').val();
+        if (selected == 1) {
+            // 主任务
+            $('#parent-task').show();
+            $('#child-task').hide();
+        } else {
+            // 子任务
+            $('#parent-task').hide();
+            $('#child-task').show();
+        }
+    }
+
     var $uiForm = $('.ui.form');
     registerSelectFormValidation("selectProtocol", $uiForm, $('#protocol'), 'protocol');
     $($uiForm).form(
@@ -299,19 +361,6 @@
                             }
                         ]
                     },
-                    spec: {
-                        identifier  : 'spec',
-                        rules: [
-                            {
-                                type   : 'empty',
-                                prompt : '请输入crontab格式表达式'
-                            },
-                            {
-                                type : 'maxLength[64]',
-                                prompt : '长度不能超过64'
-                            }
-                        ]
-                    },
                     command: {
                         identifier  : 'command',
                         rules: [