修复map加锁无效

当任务较多时, 频繁读写数据库,数据库压力大, 计划拆分为独立项目, 用Redis实现持久化

BREAKING CHANGE: 不再支持延迟任务

.gitignore

@@ -34,4 +34,7 @@ public/resource/javascript/vue.js
 gocron
 gocron.exe
 gocron-node
-gocron-node.exe
+gocron-node.exe
+
+node_modules
+package.json

.travis.yml

@@ -1,4 +1,8 @@
 language: go
 go:
   - 1.7.x
-script: go test `go list ./... | grep -v vendor`
+script: go test `go list ./... | grep -v vendor`
+
+notifications:
+  on_success: never
+  on_failure: always

README.md

@@ -9,16 +9,17 @@
 * crontab时间表达式，精确到秒
 * 任务执行失败重试设置
 * 任务超时设置
-* 延时任务
+* 任务依赖配置
 * 任务类型
     * shell任务
-    > 在远程服务器上执行shell命令, 调度器与任务执行器保持长连接  
+    > 在远程服务器上执行shell命令
     * HTTP任务
     > 访问指定的URL地址
 * 查看任务执行日志
 * 任务执行结果通知, 支持邮件、Slack
 
 ### 截图
+![流程图](https://raw.githubusercontent.com/ouqiang/gocron/master/scheduler.png)
 ![任务](https://raw.githubusercontent.com/ouqiang/gocron/master/screenshot_task.png)
 ![Slack](https://raw.githubusercontent.com/ouqiang/gocron/master/screenshot_slack.png)
     
@@ -43,15 +44,16 @@
 ## 安装
 
 ###  二进制安装
-1. 解压压缩包    
+> Windows平台默认后台运行)
+1. 解压压缩包   
 2. `cd 解压目录`   
-3. 启动
-    * 调度器启动
-        * Windows:  `gocron.exe web`            
-        * Linux、Mac OS:  `./gocron web`
-    * 任务执行器启动
-        * Windows:  `gocron-node.exe ip:port (默认0.0.0.0:5921)`            
-        * Linux、Mac OS:  `./gocron-node ip:port (默认0.0.0.0:5921)`   
+3. 启动        
+* 调度器启动        
+  * Windows: `gocron.exe web`   
+  * Linux、Mac OS:  `./gocron web`
+* 任务执行器启动
+  * Windows:  `gocron-node.exe ip:port (默认0.0.0.0:5921)`            
+  * Linux、Mac OS:  `./gocron-node ip:port (默认0.0.0.0:5921)`   
 4. 浏览器访问 http://localhost:5920
 ### 源码安装
 1. `go`语言版本1.7+
@@ -64,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) {
@@ -115,27 +121,6 @@ func initModule()  {
     // 初始化定时任务
     serviceTask := new(service.Task)
     serviceTask.Initialize()
-
-    // 初始化延时任务
-    delayTaskEnabled, err := config.Key("delay.task.enable").Bool()
-    if err != nil {
-        return
-    }
-    if !delayTaskEnabled {
-        return
-    }
-    delayTaskSlots, err := config.Key("delay.task.slots").Int()
-    if err != nil {
-        return
-    }
-    delayTaskTick := config.Key("delay.task.tick").String()
-    tick, err := time.ParseDuration(delayTaskTick)
-    if err != nil {
-        return
-    }
-
-    serviceDelayTask := new(service.DelayTask)
-    serviceDelayTask.Initialize(tick, delayTaskSlots)
 }
 
 // 解析端口
@@ -151,6 +136,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") {
@@ -215,12 +208,6 @@ func shutdown()  {
     // 停止所有任务调度
     logger.Info("停止定时任务调度")
     serviceTask.StopAll()
-    delayTaskEnable, _ := app.Setting.Key("delay.task.enable").Bool()
-    if delayTaskEnable {
-        logger.Info("停止延时任务调度")
-        serviceDelayTask := new(service.DelayTask)
-        serviceDelayTask.Stop()
-    }
     // 释放gRPC连接池
     grpcpool.Pool.ReleaseAll()
 

gocron-node.go

@@ -15,5 +15,7 @@ func main()  {
 	} else {
         addr = os.Args[1]
     }
-	server.Start(addr)
+	for {
+		server.Start(addr)
+	}
 }

gocron.go

@@ -10,7 +10,7 @@ import (
     "github.com/ouqiang/gocron/cmd"
 )
 
-const AppVersion = "0.4"
+const AppVersion = "1.0"
 
 func main() {
     app := cli.NewApp()

models/delay_task.go

@@ -1,85 +0,0 @@
-package models
-
-import (
-    "time"
-    "github.com/go-xorm/xorm"
-)
-
-
-// 延迟任务
-type DelayTask struct {
-    Id       int64     `xorm:"bigint pk autoincr"`
-    Url      string    `xorm:"varchar(128) not null"`
-    Params   string    `xorm:"varchar(256) not null default '' "`
-    Delay  int       `xorm:"mediumint notnull default 0"` // 延迟时间
-    Status   Status    `xorm:"tinyint notnull index(u_status_created) default 5"` // 状态 0:执行失败 1:执行中 2:执行成功 5: 待执行
-    Created  time.Time `xorm:"datetime notnull created index(u_status_created)"`
-    Updated  time.Time `xorm:"datetime updated"`
-    BaseModel `xorm:"-"`
-}
-
-func (task *DelayTask) Create() (insertId int64, err error) {
-    _, err = Db.Insert(task)
-    if err == nil {
-        insertId = task.Id
-    }
-
-    return
-}
-
-// 获取所有待执行任务
-func (task *DelayTask) ActiveList(endTime time.Time) ([]DelayTask, error) {
-    list := make([]DelayTask, 0)
-    fields := "id,url,params,delay,created"
-    err := Db.Where("status IN (?, ?) AND created <= ?", Waiting, Running, endTime.Format(DefaultTimeFormat)).Cols(fields).Limit(task.PageSize, task.pageLimitOffset()).Find(&list)
-
-    return list, err
-}
-
-// 获取待执行任务数量
-func (task *DelayTask) ActiveNum(endTime time.Time) (int, error) {
-    count ,err := Db.Where("status IN (?, ?) AND created <= ?", Waiting, Running, endTime.Format(DefaultTimeFormat)).Count(task)
-
-    return int(count), err
-}
-
-func (task *DelayTask) List(params CommonMap) ([]DelayTask, error) {
-    task.parsePageAndPageSize(params)
-    list := make([]DelayTask, 0)
-    session := Db.Desc("id")
-    task.parseWhere(session, params)
-    err := session.Limit(task.PageSize, task.pageLimitOffset()).Find(&list)
-
-    return list, err
-}
-
-
-// 更新任务状态
-func (task *DelayTask) UpdateStatus(id int64, status Status) (int64, error) {
-    return Db.Table(task).Id(id).Update(CommonMap{
-        "status": status,
-    })
-}
-
-// 解析where
-func (task *DelayTask) parseWhere(session *xorm.Session, params CommonMap)  {
-    if len(params) == 0 {
-        return
-    }
-    status, ok := params["Status"]
-    if ok && status.(int) > -1 {
-        session.And("status = ?", status)
-    }
-}
-
-// 删除N个月前的日志
-func (task *DelayTask) Remove(id int) (int64, error) {
-    t := time.Now().AddDate(0, -id, 0)
-    return Db.Where("created <= ?", t.Format(DefaultTimeFormat)).Delete(task)
-}
-
-func (task *DelayTask) Total(params CommonMap) (int64, error) {
-    session := Db.NewSession()
-    task.parseWhere(session, params)
-    return session.Count(task)
-}

models/migration.go

@@ -15,7 +15,7 @@ func (migration *Migration) Exec(dbName string) error {
     setting := new(Setting)
     task := new(Task)
     tables := []interface{}{
-        &User{}, task, &DelayTask{}, &TaskLog{}, &Host{}, setting,&LoginLog{},
+        &User{}, task, &TaskLog{}, &Host{}, setting,&LoginLog{},
     }
     for _, table := range tables {
         exist, err:= Db.IsTableExist(table)

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
 }
@@ -138,4 +140,12 @@ func getDbConfig() map[string]string {
     db["max_open_conns"] = app.Setting.Key("db.max.open.conns").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)

modules/rpc/client/client.go

@@ -16,11 +16,11 @@ var (
     errUnavailable = errors.New("无法连接远程服务器")
 )
 
-func Exec(ip string, port int, taskReq *pb.TaskRequest) (string, error)  {
+func ExecWithRetry(ip string, port int, taskReq *pb.TaskRequest) (string, error)  {
     tryTimes := 60
     i := 0
     for i < tryTimes {
-        output, err := exec(ip, port, taskReq)
+        output, err := Exec(ip, port, taskReq)
         if err != errUnavailable {
             return output, err
         }
@@ -31,7 +31,7 @@ func Exec(ip string, port int, taskReq *pb.TaskRequest) (string, error)  {
     return "", errUnavailable
 }
 
-func exec(ip string, port int, taskReq *pb.TaskRequest) (string, error)  {
+func Exec(ip string, port int, taskReq *pb.TaskRequest) (string, error)  {
     defer func() {
        if err := recover(); err != nil {
            logger.Error("panic#rpc/client.go:Exec#", err)
@@ -69,7 +69,7 @@ func exec(ip string, port int, taskReq *pb.TaskRequest) (string, error)  {
 
 func parseGRPCError(err error, conn *grpc.ClientConn, connClosed *bool) (string, error) {
     switch grpc.Code(err) {
-        case codes.Unavailable:
+        case codes.Unavailable, codes.Internal:
             conn.Close()
             *connClosed = true
             return "", errUnavailable

modules/rpc/grpcpool/grpc_pool.go

@@ -32,8 +32,8 @@ type GRPCPool struct {
 
 func (p *GRPCPool) Get(addr string) (*grpc.ClientConn, error)  {
     p.RLock()
-    p.RUnlock()
     pool, ok := p.conns[addr]
+    p.RUnlock()
     if !ok {
         err := p.newCommonPool(addr)
         if err != nil {
@@ -41,7 +41,9 @@ func (p *GRPCPool) Get(addr string) (*grpc.ClientConn, error)  {
         }
     }
 
+    p.RLock()
     pool = p.conns[addr]
+    p.RUnlock()
     conn, err := pool.Get()
     if err != nil {
         return nil, err

modules/rpc/server/server.go

@@ -30,6 +30,11 @@ func (s Server) Run(ctx context.Context, req *pb.TaskRequest) (*pb.TaskResponse,
 }
 
 func Start(addr string)  {
+    defer func() {
+       if err := recover(); err != nil {
+           grpclog.Println("panic", err)
+       }
+    } ()
     l, err := net.Listen("tcp", addr)
     if err != nil {
         grpclog.Fatal(err)

public/resource/javascript/main.js

@@ -26,7 +26,7 @@ function Util() {
     // ajax错误处理
     util.ajaxFailure = function() {
         // todo 错误处理
-        swal(FAILURE_MESSAGE, '未知错误', 'error');
+        swal(FAILURE_MESSAGE, '操作失败', 'error');
     };
     // get请求
     util.get = function(url, callback) {

routers/delaytask/delay_task.go

@@ -1,110 +0,0 @@
-package delaytask
-
-import (
-    "gopkg.in/macaron.v1"
-    "github.com/ouqiang/gocron/models"
-    "github.com/ouqiang/gocron/modules/utils"
-    "strings"
-    "github.com/ouqiang/gocron/service"
-    "github.com/ouqiang/gocron/modules/logger"
-    "github.com/Unknwon/paginater"
-    "fmt"
-    "github.com/ouqiang/gocron/routers/base"
-    "html/template"
-    "github.com/ouqiang/gocron/modules/app"
-)
-
-func Index(ctx *macaron.Context)  {
-    delayTaskModel := new(models.DelayTask)
-    queryParams := parseQueryParams(ctx)
-    total, err := delayTaskModel.Total(queryParams)
-    tasks, err := delayTaskModel.List(queryParams)
-    if err != nil {
-        logger.Error(err)
-    }
-    PageParams := fmt.Sprintf("status=%d&page_size=%d",
-        queryParams["Status"], queryParams["PageSize"]);
-    queryParams["PageParams"] = template.URL(PageParams)
-    p := paginater.New(int(total), queryParams["PageSize"].(int), queryParams["Page"].(int), 5)
-    ctx.Data["Pagination"] = p
-    ctx.Data["Title"] = "延时任务列表"
-    ctx.Data["Tasks"] = tasks
-    ctx.Data["Params"] = queryParams
-    ctx.HTML(200, "task/delay_task")
-}
-
-func Create(ctx *macaron.Context) string {
-    url := ctx.QueryTrim("url")
-    params := ctx.QueryTrim("params")
-    delay := ctx.QueryInt("delay")
-    json := utils.JsonResponse{}
-    delayTaskEnabled, _ := app.Setting.Key("delay.task.enable").Bool()
-    if !delayTaskEnabled {
-        return json.CommonFailure("系统未开启延时任务")
-    }
-    if url == ""  {
-        return json.CommonFailure("url地址不能为空")
-    }
-    lowerUrl := strings.ToLower(url)
-    if !strings.HasPrefix(lowerUrl, "http") &&
-        !strings.HasPrefix(lowerUrl, "https") {
-        return json.CommonFailure("无效的url地址")
-    }
-    if len(url) > 128 {
-        return json.CommonFailure("url长度不能超过128")
-    }
-    maxDelay := 1 << 31
-    if delay <= 0 || delay > maxDelay {
-        return json.CommonFailure("无效的delay, 取值范围1-(2^31-1)")
-    }
-    if len(params) > 256 {
-        return json.CommonFailure("params长度不能超过256")
-    }
-
-    delayTask := new(models.DelayTask)
-    delayTask.Url = url
-    delayTask.Params = params
-    delayTask.Delay = delay
-    delayTask.Status = models.Waiting
-    _, err := delayTask.Create()
-
-    if err != nil {
-        return json.CommonFailure("添加失败", err)
-    }
-
-    logger.Infof("新增延时任务#id-%d#url-%s#params-%s#delay-%d",
-        delayTask.Id, delayTask.Url, delayTask.Params, delayTask.Delay)
-    delayTaskService := new(service.DelayTask)
-    delayTaskService.Add(*delayTask)
-
-    return json.Success("添加成功", nil)
-}
-
-// 删除N个月前的日志
-func Remove(ctx *macaron.Context) string {
-    month := ctx.ParamsInt(":id")
-    json := utils.JsonResponse{}
-    if month < 1 || month > 12 {
-        return json.CommonFailure("参数取值范围1-12")
-    }
-    delayTaskModel := new(models.DelayTask)
-    _, err := delayTaskModel.Remove(month)
-    if err != nil {
-        return json.CommonFailure("删除失败", err)
-    }
-
-    return json.Success("删除成功", nil)
-}
-
-// 解析查询参数
-func parseQueryParams(ctx *macaron.Context) (models.CommonMap) {
-    var params models.CommonMap = models.CommonMap{}
-    status := ctx.QueryInt("status")
-    if status >=0 {
-        status -= 1
-    }
-    params["Status"] = status
-    base.ParsePageAndPageSize(ctx, params)
-
-    return params
-}

routers/host/host.go

@@ -14,6 +14,8 @@ import (
     "github.com/go-macaron/binding"
     "github.com/ouqiang/gocron/modules/rpc/grpcpool"
     "strings"
+    "github.com/ouqiang/gocron/modules/rpc/client"
+    "github.com/ouqiang/gocron/modules/rpc/proto"
 )
 
 func Index(ctx *macaron.Context)  {
@@ -160,6 +162,27 @@ func Remove(ctx *macaron.Context) string  {
     return json.Success("操作成功", nil)
 }
 
+func Ping(ctx *macaron.Context) string  {
+    id := ctx.ParamsInt(":id")
+    hostModel := new(models.Host)
+    err := hostModel.Find(id)
+    json := utils.JsonResponse{}
+    if err != nil || hostModel.Id <= 0{
+        return json.CommonFailure("主机不存在", err)
+    }
+
+
+    taskReq := &rpc.TaskRequest{}
+    taskReq.Command = "echo hello"
+    taskReq.Timeout = 10
+    output, err := client.Exec(hostModel.Name, hostModel.Port, taskReq)
+    if err != nil {
+        return json.CommonFailure("连接失败-" + err.Error() + " " + output, err)
+    }
+
+    return json.Success("连接成功", nil)
+}
+
 // 解析查询参数
 func parseQueryParams(ctx *macaron.Context) (models.CommonMap) {
     var params models.CommonMap = models.CommonMap{}

routers/install/install.go

@@ -115,9 +115,6 @@ func writeConfig(form InstallForm) error {
         "db.max.open.conns", "100",
         "allow_ips", "",
         "app.name", "定时任务管理系统", // 应用名称
-        "delay.task.enable", "false", // 是否开启延时任务
-        "delay.task.slots", "3600", // 时间轮槽数量
-        "delay.task.tick", "1s", // 时间轮每次转动的时间
         "api.key", "",
         "api.secret", "",
     }

routers/routers.go

@@ -17,7 +17,6 @@ import (
     "github.com/go-macaron/gzip"
     "github.com/ouqiang/gocron/routers/manage"
     "github.com/ouqiang/gocron/routers/loginlog"
-    "github.com/ouqiang/gocron/routers/delaytask"
     "time"
     "strconv"
 )
@@ -60,17 +59,13 @@ func Register(m *macaron.Macaron) {
         m.Get("/run/:id", task.Run)
     })
 
-    // 延时任务
-    m.Group("/delaytask", func() {
-        m.Get("", delaytask.Index)
-    })
-
     // 主机
     m.Group("/host", func() {
         m.Get("/create", host.Create)
         m.Get("/edit/:id", host.Edit)
         m.Post("/store", binding.Bind(host.HostForm{}), host.Store)
         m.Get("", host.Index)
+        m.Get("/ping/:id", host.Ping)
         m.Post("/remove/:id", host.Remove)
     })
 
@@ -97,8 +92,8 @@ func Register(m *macaron.Macaron) {
     // API
     m.Group("/api/v1", func() {
        m.Post("/tasklog/remove/:id", tasklog.Remove)
-       m.Post("/delaytask/push", delaytask.Create)
-       m.Post("/delaytask/log/remove/:id", delaytask.Remove)
+       m.Post("/task/enable/:id", task.Enable)
+       m.Post("/task/disable/:id", task.Disable)
     }, apiAuth);
 
     // 404错误

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/delay_task.go

@@ -1,133 +0,0 @@
-package service
-
-import (
-    "github.com/ouqiang/gocron/models"
-    "time"
-    "github.com/ouqiang/gocron/modules/logger"
-    "math"
-    "github.com/ouqiang/gocron/modules/httpclient"
-    "strings"
-    "github.com/ouqiang/timewheel"
-    "fmt"
-    "github.com/ouqiang/gocron/modules/app"
-)
-
-var tw  *timewheel.TimeWheel
-
-type DelayTask struct {}
-
-// 从数据库中取出所有延迟任务
-func (task *DelayTask) Initialize(tick time.Duration, slots int)  {
-    tw = timewheel.New(tick, slots, task.Run)
-    tw.Start()
-    taskModel := new(models.DelayTask)
-    currentTime := time.Now()
-    taskNum, err := taskModel.ActiveNum(currentTime)
-    if err != nil {
-        logger.Error("延迟任务初始化#获取待执行的任务失败", err)
-        return
-    }
-    if taskNum == 0 {
-        logger.Debugf("延迟任务初始化#待执行的任务数量为0")
-        return
-    }
-    pageSize := 100
-    totalPage := int( math.Ceil(float64(taskNum) / float64(pageSize)) )
-    logger.Infof("延迟任务初始化#待执行的任务数量-%d#共%d页#每页取%d条", taskNum, totalPage, pageSize)
-    taskModel.PageSize = pageSize
-    for page := 1; page <= totalPage; page++ {
-        taskModel.Page = page
-        logger.Debugf("延迟任务初始化#取出任务列表#第%d页", page)
-        taskList, err := taskModel.ActiveList(currentTime)
-        if err != nil {
-            logger.Error("延迟任务初始化#获取任务列表失败", err)
-        }
-        task.BatchAdd(taskList)
-    }
-    logger.Info("延迟任务初始化完成")
-}
-
-// 批量添加任务
-func (task *DelayTask) BatchAdd(taskList []models.DelayTask)  {
-    for _, item := range(taskList) {
-        task.Add(item)
-    }
-}
-
-// 添加任务
-func (task *DelayTask) Add(taskModel models.DelayTask)  {
-    currentTimestamp := time.Now().Unix()
-    execTimestamp := taskModel.Created.Unix() + int64(taskModel.Delay)
-    // 时间过期, 立即执行任务
-    data := []interface{}{taskModel.Id, taskModel.Url, taskModel.Params}
-    if execTimestamp <= currentTimestamp {
-        go task.Run(data)
-        return
-    }
-    delay := execTimestamp - currentTimestamp
-    tw.Add(time.Duration(delay) * time.Second, data)
-}
-
-// 运行任务
-func (task *DelayTask) Run(data []interface{})  {
-    if len(data) < 3 {
-        logger.Errorf("延时任务开始执行#参数不足#%+v", data)
-        return
-    }
-    id := data[0].(int64)
-    url := data[1].(string)
-    params := data[2].(string)
-    if id <= 0 || url == "" {
-        logger.Errorf("延时任务开始执行#参数为空#%+v", data)
-        return
-    }
-    taskModel := new(models.DelayTask)
-    _, err := taskModel.UpdateStatus(id, models.Running)
-    if err != nil {
-        logger.Error("延迟任务开始执行#更新任务状态失败", err)
-        return
-    }
-    timeout := 300
-    tryTimes := 3
-    success := false
-    logger.Infof("延迟任务开始执行#id-%d#url-%s#params-%s", id, url, params)
-    for i := 0; i < tryTimes; {
-        response := httpclient.PostParams(url, params, timeout)
-        if response.StatusCode == 200 && strings.TrimSpace(response.Body) == "success"{
-            success = true
-            break;
-        }
-        i++
-        if i < tryTimes {
-            msg := fmt.Sprintf("延迟任务执行失败#重试第%d次#任务Id-%d#HTTP状态码-%d#HTTP-BODY-%s",
-                i,id,response.StatusCode,response.Body)
-            logger.Error(msg)
-            FailureNotify(msg)
-            time.Sleep(30 * time.Second)
-        }
-    }
-    logger.Infof("延迟任务执行完成#id-%d", id)
-    var status models.Status
-    if success {
-        status = models.Finish
-    } else {
-        status = models.Failure
-    }
-    _ ,err = taskModel.UpdateStatus(id, status)
-    if err != nil {
-        logger.Error("延迟任务执行完成#更新任务状态失败", err)
-    }
-}
-
-func (task *DelayTask) Stop()  {
-    tw.Stop()
-}
-
-func FailureNotify(message string)  {
-    notifyUrl := app.Setting.Key("delay.task.failure.notify.url").String()
-    notifyUrl = strings.TrimSpace(notifyUrl)
-    if notifyUrl != "" {
-        params := fmt.Sprintf("error=%s", message)
-        httpclient.PostParams(notifyUrl, params, 60)
-    }
-}

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)
@@ -164,7 +176,7 @@ func (h *RPCHandler) Run(taskModel models.TaskHost) (result string, err error)
     taskRequest.Timeout = int32(taskModel.Timeout)
     taskRequest.Command = taskModel.Command
 
-    return rpcClient.Exec(taskModel.Name, taskModel.Port, taskRequest)
+    return rpcClient.ExecWithRetry(taskModel.Name, taskModel.Port, taskRequest)
 }
 
 
@@ -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/host/index.html

@@ -53,6 +53,7 @@
                     <button class="ui positive button" onclick="util.removeConfirm('/host/remove/{{{.Id}}}')">删除</button><br>
                     <div style="margin-top: 5px;">
                         <a class="ui twitter button" href="/task?host_id={{{.Id}}}">查看任务</a>
+                        <button class="ui blue button" @click="ping({{{.Id}}})">连接测试</button>
                     </div>
                 </td>
             </tr>
@@ -63,4 +64,17 @@
     </div>
 </div>
 
+<script type="text/javascript">
+    var Vue = new Vue({
+        el: '.ui.striped.table',
+        methods: {
+            ping: function(id) {
+                util.get("/host/ping/" + id, function(code, message) {
+                    swal('操作成功', '连接成功', 'success');
+                })
+            }
+        }
+    });
+</script>
+
 {{{ template "common/footer" . }}}

templates/install/create.html

@@ -90,12 +90,6 @@
     $('.ui.form').form(
             {
                 onSuccess: function(event, fields) {
-                    swal({
-                        title: '',
-                        text: "系统安装中.......",
-                        type: 'info',
-                        showConfirmButton: false
-                    });
                     util.post('/install/store', fields, function(code, message) {
                         swal('安装成功');
                         setTimeout(function() {

templates/task/delay_task.html

@@ -1,84 +0,0 @@
-{{{ template "common/header" . }}}
-<style type="text/css">
-  pre {
-    white-space: pre-wrap;
-    word-wrap: break-word;
-    padding:10px;
-    background-color: #4C4C4C;
-    color: white;
-  }
-</style>
-<div class="ui grid">
-  <!--the vertical menu-->
-  {{{ template "task/menu" . }}}
-
-  <div class="twelve wide column">
-    <div class="pageHeader">
-      <div class="segment">
-        <h3 class="ui dividing header">
-          <div class="content">
-          </div>
-        </h3>
-      </div>
-    </div>
-    <form class="ui form">
-      <div class="fields search">
-        <div class="field">
-          <select name="status">
-            <option value="0">任务状态</option>
-            <option value="1" {{{if eq .Params.Status 0}}}selected{{{end}}} >失败</option>
-            <option value="2" {{{if eq .Params.Status 1}}}selected{{{end}}}>执行中</option>
-            <option value="3" {{{if eq .Params.Status 2}}}selected{{{end}}}>成功</option>
-            <option value="6" {{{if eq .Params.Status 5}}}selected{{{end}}}>待执行</option>
-          </select>
-        </div>
-        <div class="field">
-          <button class="ui linkedin submit button">搜索</button>
-        </div>
-      </div>
-    </form>
-    <table class="ui pink table">
-      <thead>
-      <tr>
-        <th>任务ID</th>
-        <th>URL</th>
-        <th>参数</th>
-        <th>延迟时间</th>
-        <th>创建时间</th>
-        <th>完成时间</th>
-        <th>状态</th>
-      </tr>
-      </thead>
-      <tbody>
-      {{{range $i, $v := .Tasks}}}
-      <tr>
-        <td>{{{.Id}}}</td>
-        <td>{{{.Url}}}</td>
-        <td>{{{.Params}}}</td>
-        <td>{{{.Delay}}}秒</td>
-        <td>{{{.Created.Format "2006-01-02 15:04:05" }}}</td>
-        <td>
-          {{{if or (eq .Status 0) (eq .Status 2) }}}
-            {{{.Updated.Format "2006-01-02 15:04:05" }}}
-          {{{end}}}
-        </td>
-        <td>
-          {{{if eq .Status 2}}}
-          成功
-          {{{else if eq .Status 1}}}
-          <span style="color:green">执行中</span>
-          {{{else if eq .Status 0}}}
-          <span style="color:red">失败</span>
-          {{{else if eq .Status 5}}}
-          <span style="color:#43A102">待执行</span>
-          {{{end}}}
-        </td>
-      </tr>
-      {{{end}}}
-      </tbody>
-    </table>
-    {{{ template "common/pagination" .}}}
-  </div>
-</div>
-
-{{{ template "common/footer" . }}}

templates/task/index.html

@@ -54,6 +54,7 @@
             <tr>
                 <th>任务ID</th>
                 <th>任务名称</th>
+                <th>任务类型</th>
                 <th>cron表达式</th>
                 <th>执行方式</th>
                 <th>超时时间</th>
@@ -69,20 +70,27 @@
                     <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 .Status 1}}}
-                                <button class="ui primary button" @click="changeStatus({{{.Id}}},{{{.Status}}})">停止</button>
-                                {{{else}}}
-                                <button class="ui blue button" @click="changeStatus({{{.Id}}},{{{.Status}}})">激活 </button>
+                                {{{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>
@@ -97,7 +105,6 @@
     </div>
 </div>
 
-
 <script type="text/javascript">
     $('.ui.checkbox').checkbox();
 

templates/task/menu.html

@@ -7,9 +7,6 @@
             <a class="item {{{if eq .URI "/task/log"}}}active teal{{{end}}} " href="/task/log">
                 <i class="bar chart icon"></i> 定时任务日志
             </a>
-            <a class="item {{{if eq .URI "/delaytask"}}}active teal{{{end}}} " href="/delaytask">
-                <i class="bar chart icon"></i> 延时任务日志
-            </a>
         </div>
     </div>
 </div>

templates/task/task_form.html

@@ -19,19 +19,63 @@
                     <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">
-                        crontab表达式
+                    <div class="content">任务类型</div>
+                    <div class="ui message">
+                        主任务可以配置多个子任务, 当主任务执行完成后，自动执行子任务<br>
+                        任务类型新增后不能变更
                     </div>
                 </label>
-                <div class="ui small input">
-                    <input type="text" name="spec" value="{{{.Task.Spec}}}" placeholder="秒 分 时 天 月 周"/>
+                <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 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>
+            <div class="two fields">
+                <div class="field">
+                    <label>
+                        <div class="content">
+                            crontab表达式
+                        </div>
+                    </label>
+                    <div class="ui small input">
+                        <input type="text" name="spec" value="{{{.Task.Spec}}}" placeholder="秒 分 时 天 月 周"/>
+                    </div>
                 </div>
             </div>
         </div>
@@ -46,7 +90,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}}}
@@ -62,19 +106,15 @@
                 <textarea rows="5" name="command" placeholder="请输入系统命令" id="command">{{{.Task.Command}}}</textarea>
             </div>
         </div>
-        <div class="six fields">
+        <div class="three 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>
-        <div class="three fields">
             <div class="field">
                 <label>允许多实例同时运行</label>
                 <select name="multi">
@@ -82,6 +122,7 @@
                     <option value="1"{{{if .Task}}} {{{if eq .Task.Multi 1}}}selected{{{end}}} {{{end}}}>是</option>
                 </select>
             </div>
+
         </div>
         <div class="three fields">
             <div class="field">
@@ -107,7 +148,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 +185,7 @@
 <script type="text/javascript">
     $(function() {
         changeCommandPlaceholder();
+        changeLevel();
         changeProtocol();
         showNotify();
     });
@@ -153,7 +195,9 @@
         changeProtocol();
     });
 
-
+    $('#level').change(function() {
+        changeLevel();
+    });
 
     $('#task-status').change(function() {
         var selected = $(this).val();
@@ -269,6 +313,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 +356,6 @@
                             }
                         ]
                     },
-                    spec: {
-                        identifier  : 'spec',
-                        rules: [
-                            {
-                                type   : 'empty',
-                                prompt : '请输入crontab格式表达式'
-                            },
-                            {
-                                type : 'maxLength[64]',
-                                prompt : '长度不能超过64'
-                            }
-                        ]
-                    },
                     command: {
                         identifier  : 'command',
                         rules: [

vendor/github.com/ouqiang/timewheel/LICENSE

@@ -1,21 +0,0 @@
-MIT License
-
-Copyright (c) 2017 qiang.ou
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.

vendor/github.com/ouqiang/timewheel/README.md

@@ -1,40 +0,0 @@
-# timewheel
-Golang实现的时间轮
-
-
-![时间轮](https://raw.githubusercontent.com/ouqiang/timewheel/master/timewheel.jpg)
-
-# 安装
-
-```shell
-go get -u github.com/ouqiang/timewheel
-```
-
-# 使用
-
-```go
-package main
-
-import (
-    "github.com/ouqiang/timewheel"
-    "time"
-)
-
-func main()  {
-    // tick刻度为1秒, 3600个槽
-    tw := timewheel.New(1 * time.Second, 3600)
-    tw.Start()
-    tw.Add(5 * time.Second, func() {
-        // do something
-    })
-    tw.Add(10 * time.Minute, func() {
-        // do something
-    })
-    tw.Add(35 * time.Hour, func() {
-        // do something
-    })
-    // 停止
-    tw.Stop()
-}
-```
-

vendor/github.com/ouqiang/timewheel/timewheel.go

@@ -1,126 +0,0 @@
-package timewheel
-
-import (
-    "time"
-    "container/list"
-)
-
-// @author qiang.ou<qingqianludao@gmail.com>
-
-type Job func([]interface{})
-
-type TimeWheel struct {
-    interval time.Duration
-    ticker *time.Ticker
-    slots []*list.List
-    currentPos int
-    slotNum int
-    job Job
-    taskChannel chan Task
-    stopChannel chan bool
-}
-
-
-type Task struct {
-    delay time.Duration
-    circle int
-    data []interface{}
-}
-
-func New(interval time.Duration, slotNum int, job Job) *TimeWheel {
-    if  interval <= 0 || slotNum <= 0 || job == nil {
-        return nil
-    }
-    tw := &TimeWheel{
-        interval: interval,
-        slots: make([]*list.List, slotNum),
-        currentPos: 0,
-        job: job,
-        slotNum: slotNum,
-        taskChannel: make(chan Task),
-        stopChannel: make(chan bool),
-    }
-
-    tw.initSlots()
-
-    return tw
-}
-
-func (tw *TimeWheel) initSlots()  {
-    for i := 0; i < tw.slotNum; i++ {
-        tw.slots[i] = list.New()
-    }
-}
-
-func (tw *TimeWheel) Start()  {
-    tw.ticker = time.NewTicker(tw.interval)
-    go tw.start()
-}
-
-func (tw *TimeWheel) Add(delay time.Duration, data []interface{})  {
-    if delay <= 0  {
-        return
-    }
-    tw.taskChannel <- Task{delay:delay, data: data}
-}
-
-func (tw *TimeWheel) Stop()  {
-    tw.stopChannel <- true
-}
-
-func (tw *TimeWheel) start()  {
-    for {
-        select {
-        case <- tw.ticker.C:
-            tw.tickHandler()
-        case task := <- tw.taskChannel:
-            tw.addTask(&task)
-        case <- tw.stopChannel:
-            tw.ticker.Stop()
-            return
-        }
-    }
-}
-
-func (tw *TimeWheel) tickHandler()  {
-    l := tw.slots[tw.currentPos]
-    tw.scanAndRunTask(l)
-    if tw.currentPos == tw.slotNum - 1 {
-        tw.currentPos = 0
-    } else {
-        tw.currentPos++
-    }
-}
-
-func (tw *TimeWheel) scanAndRunTask(l *list.List)  {
-    for e := l.Front(); e != nil; {
-        task := e.Value.(*Task)
-        if task.circle > 0 {
-            task.circle--
-            e = e.Next()
-            continue
-        }
-
-        go tw.job(task.data)
-        next := e.Next()
-        l.Remove(e)
-        e = next
-    }
-}
-
-func (tw *TimeWheel) addTask(task *Task)  {
-    pos, circle := tw.getPositionAndCircle(task.delay)
-    task.circle = circle
-
-    tw.slots[pos].PushBack(task)
-}
-
-func (tw *TimeWheel) getPositionAndCircle(d time.Duration) (pos int, circle int) {
-    delaySeconds := int(d.Seconds())
-    intervalSeconds := int(tw.interval.Seconds())
-    circle = int(delaySeconds / intervalSeconds /  tw.slotNum)
-    pos = int(tw.currentPos + delaySeconds / intervalSeconds) % tw.slotNum
-
-
-    return
-}

vendor/vendor.json

@@ -142,12 +142,6 @@
 			"revision": "09cded8978dc9e80714c4d85b0322337b0a1e5e0",
 			"revisionTime": "2016-03-02T07:53:16Z"
 		},
-		{
-			"checksumSHA1": "kIFW+u9fHefC8sWE4W9pYIfJv5k=",
-			"path": "github.com/ouqiang/timewheel",
-			"revision": "c28ec761087c32fd75ad7514db2a4988d5c872d9",
-			"revisionTime": "2017-05-14T12:16:09Z"
-		},
 		{
 			"checksumSHA1": "cVGA2CJTJsCAVa5VKTM8k/ma/BU=",
 			"path": "github.com/silenceper/pool",