feat($task): 支持任务同时在多个节点上运行

Closes #7

.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

@@ -2,23 +2,26 @@
 # gocron - 定时任务管理系统
 
 # 项目简介
-使用Go语言开发的定时任务集中调度和管理系统, 用于替代Linux-crontab [查看文档](https://github.com/ouqiang/gocron/wiki)
+使用Go语言开发的定时任务集中调度和管理系统, 用于替代Linux-crontab [查看文档](https://github.com/ouqiang/gocron/wiki)  
+
+原有的延时任务拆分为独立项目[延迟队列](https://github.com/ouqiang/delay-queue)  
 
 ## 功能特性
-* Web界面管理定时任务, 支持动态添加、删除、编辑任务
+* Web界面管理定时任务, 支持动态添加、删除任务
 * crontab时间表达式，精确到秒
 * 任务执行失败重试设置
 * 任务超时设置
-* 延时任务
+* 任务依赖配置
 * 任务类型
     * shell任务
-    > 在远程服务器上执行shell命令, 调度器与任务执行器保持长连接  
+    > 在任务节点上执行shell命令, 支持任务同时在多个节点上运行
     * HTTP任务
-    > 访问指定的URL地址
+    > 访问指定的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)
     
@@ -30,50 +33,44 @@
 
 
 ## 下载
-* 调度器(管理后台)
-    * [Linux-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-linux-amd64.tar.gz)
-    * [Mac OS-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-darwin-amd64.tar.gz)
-    * [Windows-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-windows-amd64.zip)
-* 任务执行器(安装在远程主机上, 执行shell命令需安装)
-     * [Linux-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-node-linux-amd64.tar.gz)
-     * [Mac OS-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-node-darwin-amd64.tar.gz)
-     * [Windows-64位](http://opns468ov.bkt.clouddn.com/gocron/gocron-node-windows-amd64.zip)
+[v1.1](https://github.com/ouqiang/gocron/releases/tag/v1.1)
 
 
 ## 安装
 
 ###  二进制安装
-1. 解压压缩包    
+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`
+* 任务节点启动, 默认监听0.0.0.0:5921
+  * Windows:  `gocron-node.exe`
+  * Linux、Mac OS:  `./gocron-node`
 4. 浏览器访问 http://localhost:5920
+
 ### 源码安装
 1. `go`语言版本1.7+
 2. `go get -d github.com/ouqiang/gocron`
 3. 编译 
     * 调度器 `go build -tags gocron -o gocron`
-    * 任务执行器 `go build -tags node -o gocron-node`
+    * 任务节点 `go build -tags node -o gocron-node`
 4. 启动、访问方式同上
 
 ### 命令
 
 * gocron web
+    * --host 默认0.0.0.0
     * -p 端口, 指定端口, 默认5920
     * -e 指定运行环境, dev|test|prod, dev模式下可查看更多日志信息, 默认prod
-    * -d 后台运行
     * -h 查看帮助
-* gocron serv 
-    * -s stop|status stop:停止gocron status:查看运行状态
-* gocron-node ip:port, 默认0.0.0.0:5921 
+* gocron-node
+    *  -allow-root *nix平台允许以root用户运行
+    * -s ip:port 监听地址
 
 ## 程序使用的组件
-* web框架 [Macaron](http://go-macaron.com/)
+* Web框架 [Macaron](http://go-macaron.com/)
 * 定时任务调度 [Cron](https://github.com/robfig/cron)
 * ORM [Xorm](https://github.com/go-xorm/xorm)
 * UI框架 [Semantic UI](https://semantic-ui.com/)
@@ -82,3 +79,14 @@
 
 ## 反馈
 提交[issue](https://github.com/ouqiang/gocron/issues/new)
+
+## ChangeLog
+
+v1.1
+--------
+
+* 任务可同时在多个节点上运行
+* *nix平台默认禁止以root用户运行任务节点
+* 子任务命令中增加预定义占位符, 子任务可根据主任务运行结果执行相应操作
+* 删除守护进程模块
+* Web访问日志输出到终端

build.sh

@@ -54,7 +54,7 @@ fi
 
 echo '开始编译调度器'
 if [[ $OS = 'windows' ]];then
-    GOOS=$OS GOARCH=$ARCH go build -tags gocron -ldflags '-w -H windowsgui'
+    GOOS=$OS GOARCH=$ARCH go build -tags gocron -ldflags '-w'
 else
     GOOS=$OS GOARCH=$ARCH go build -tags gocron -ldflags '-w'
 fi

build_node.sh

@@ -60,7 +60,7 @@ fi
 
 echo '开始编译任务节点'
 if [[ $OS = 'windows' ]];then
-    GOOS=$OS GOARCH=$ARCH go build -tags node -ldflags '-w -H windowsgui' -o $EXEC_NAME
+    GOOS=$OS GOARCH=$ARCH go build -tags node -ldflags '-w' -o $EXEC_NAME
 else
     GOOS=$OS GOARCH=$ARCH go build -tags node -ldflags '-w' -o $EXEC_NAME
 fi

cmd/serv.go

@@ -1,67 +0,0 @@
-package cmd
-
-import (
-    "github.com/urfave/cli"
-    "fmt"
-    "github.com/ouqiang/gocron/modules/utils"
-    "github.com/ouqiang/gocron/modules/app"
-    "os"
-    "syscall"
-)
-
-var CmdServ = cli.Command{
-    Name:   "serv",
-    Usage:  "manage gocron, ./gocron serv -s stop|status",
-    Action: runServ,
-    Flags: []cli.Flag{
-        cli.StringFlag{
-            Name: "s",
-            Value:"",
-            Usage: "stop|status",
-        },
-    },
-}
-
-func runServ(ctx *cli.Context)  {
-    if utils.IsWindows() {
-        fmt.Println("not support on windows")
-        return
-    }
-    option := ctx.String("s")
-    if !utils.InStringSlice([]string{"stop", "status"}, option) {
-        fmt.Println("invalid option")
-        return
-    }
-    app.InitEnv()
-    pid := app.GetPid()
-    if pid <= 0 {
-        fmt.Println("not running")
-        return
-    }
-    process ,err := os.FindProcess(pid)
-    if err != nil {
-        fmt.Println("not running", err)
-        return
-    }
-    switch option {
-        case "stop":
-            stop(process)
-        case "status":
-            status(process)
-    }
-}
-
-func stop(process *os.Process)  {
-    fmt.Println("stopping gocron......")
-    err := process.Signal(syscall.SIGTERM)
-    if err != nil {
-        fmt.Println("failed to kill process", err)
-    } else {
-        fmt.Println("stopped")
-    }
-}
-
-
-func status(process *os.Process)  {
-    fmt.Printf("running, pid-[%d]\n", process.Pid)
-}

cmd/web.go

@@ -13,24 +13,23 @@ import (
     "github.com/ouqiang/gocron/models"
     "github.com/ouqiang/gocron/modules/setting"
     "time"
-    "io"
-    "fmt"
-    "path/filepath"
-    "os/exec"
-    "github.com/ouqiang/gocron/modules/utils"
     "github.com/ouqiang/gocron/modules/rpc/grpcpool"
 )
 
 // web服务器默认端口
 const DefaultPort = 5920
 
-const InitProcess = 1
 
 var CmdWeb = cli.Command{
     Name:   "web",
     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,
@@ -41,62 +40,27 @@ var CmdWeb = cli.Command{
             Value: "prod",
             Usage: "runtime environment, dev|test|prod",
         },
-        cli.BoolFlag{
-            Name: "d",
-            Usage: "-d=true, run as daemon process",
-        },
     },
 }
 
 func runWeb(ctx *cli.Context) {
-    // 设置守护进程
-    becomeDaemon(ctx)
     // 设置运行环境
     setEnvironment(ctx)
     // 初始化应用
     app.InitEnv()
-    app.WritePid()
     // 初始化模块 DB、定时任务等
     initModule()
     // 捕捉信号,配置热更新等
     go catchSignal()
-    m := macaron.NewWithLogger(getWebLogWriter())
+    m := macaron.Classic()
 
     // 注册路由
     routers.Register(m)
     // 注册中间件.
     routers.RegisterMiddleware(m)
+    host := parseHost(ctx)
     port := parsePort(ctx)
-    fmt.Println("server start")
-    m.Run(port)
-}
-
-func becomeDaemon(ctx *cli.Context) {
-    // 不支持windows
-    if utils.IsWindows() {
-        return
-    }
-    if !ctx.IsSet("d") {
-        return
-    }
-
-    if os.Getppid() == InitProcess {
-        // 子进程不再处理
-        return
-    }
-
-    filePath, _:= filepath.Abs(os.Args[0])
-    cmd := exec.Command(filePath, os.Args[1:]...)
-    cmd.Stdin = os.Stdin
-    cmd.Stdout = os.Stdout
-    cmd.Stderr = os.Stderr
-    err := cmd.Start()
-    if err != nil {
-        logger.Fatal("创建守护进程失败", err)
-    }
-
-    // 父进程退出, 子进程由init-1号进程收养
-    os.Exit(0)
+    m.Run(host, port)
 }
 
 func initModule()  {
@@ -115,27 +79,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 +94,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") {
@@ -184,25 +135,10 @@ func catchSignal()  {
     }
 }
 
-func getWebLogWriter() io.Writer  {
-    if macaron.Env == macaron.DEV {
-        return os.Stdout
-    }
-    logFile := app.LogDir + "/access.log"
-    var err error
-    w, err := os.OpenFile(logFile, os.O_RDWR|os.O_CREATE|os.O_APPEND ,0666)
-    if err != nil {
-        fmt.Printf("日志文件[%s]打开失败", logFile)
-        panic(err)
-    }
-
-    return w
-}
 
 // 应用退出
 func shutdown()  {
     defer func() {
-        app.RemovePid()
         logger.Info("已退出")
         os.Exit(0)
     }()
@@ -215,14 +151,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()
 
     taskNumInRunning := service.TaskNum.Num()
     logger.Infof("正在运行的任务有%d个", taskNumInRunning)
@@ -236,4 +164,7 @@ func shutdown()  {
         time.Sleep(3 * time.Second)
         taskNumInRunning = service.TaskNum.Num()
     }
+
+    // 释放gRPC连接池
+    grpcpool.Pool.ReleaseAll()
 }

gocron-node.go

@@ -5,15 +5,33 @@ package main
 
 import (
 	"github.com/ouqiang/gocron/modules/rpc/server"
-	"os"
+    "flag"
+    "runtime"
+    "os"
+    "fmt"
 )
 
+const AppVersion = "1.1"
+
 func main()  {
-	var addr string
-	if (len(os.Args) < 2) {
-		addr = "0.0.0.0:5921"
-	} else {
-        addr = os.Args[1]
+	var serverAddr string
+    var allowRoot bool
+    var version bool
+    flag.BoolVar(&allowRoot, "allow-root", false, "./gocron-node -allow-root")
+    flag.StringVar(&serverAddr, "s", "0.0.0.0:5921", "./gocron-node -s ip:port")
+    flag.BoolVar(&version, "v", false, "./gocron-node -v")
+    flag.Parse()
+
+    if version {
+        fmt.Println(AppVersion)
+        os.Exit(0)
     }
-	server.Start(addr)
+
+    if runtime.GOOS != "windows" && os.Getuid() == 0 && !allowRoot   {
+        fmt.Println("Do not run gocron-node as root user")
+        os.Exit(1)
+    }
+
+
+	server.Start(serverAddr)
 }

gocron.go

@@ -10,7 +10,7 @@ import (
     "github.com/ouqiang/gocron/cmd"
 )
 
-const AppVersion = "0.4"
+const AppVersion = "1.1"
 
 func main() {
     app := cli.NewApp()
@@ -19,7 +19,6 @@ func main() {
     app.Version = AppVersion
     app.Commands = []cli.Command{
         cmd.CmdWeb,
-        cmd.CmdServ,
     }
     app.Flags = append(app.Flags, []cli.Flag{}...)
     app.Run(os.Args)

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

@@ -12,6 +12,7 @@ type Host struct {
     Port      int       `xorm:"notnull default 22"`               // 主机端口
     Remark    string    `xorm:"varchar(100) notnull default '' "` // 备注
     BaseModel       `xorm:"-"`
+    Selected bool   `xorm:"-"`
 }
 
 // 新增

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{},&TaskHost{},
     }
     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
@@ -53,8 +54,8 @@ func (model *BaseModel) parsePageAndPageSize(params CommonMap) {
     if model.Page <= 0 {
         model.Page = Page
     }
-    if model.PageSize <= 0 || model.PageSize > MaxPageSize {
-        model.PageSize = PageSize
+    if model.PageSize <= 0  {
+        model.PageSize = MaxPageSize
     }
 }
 
@@ -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,36 +14,46 @@ 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，
     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:"-"`
+    Hosts []TaskHostDetail `xorm:"-"`
 }
 
-type TaskHost struct {
-    Task `xorm:"extends"`
-    Name string
-    Port int
-    Alias string
-}
-
-func (TaskHost) TableName() string  {
-    return TablePrefix + "task"
+func taskHostTableName() []string {
+    return []string{TablePrefix + "task_host", "th"}
 }
 
 // 新增
@@ -59,6 +70,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 +80,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,remark,notify_status,notify_type,notify_receiver_id, dependency_task_id, dependency_status").
+    Update(task)
 }
 
 // 更新
@@ -92,28 +106,48 @@ 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)
+func (task *Task) ActiveList() ([]Task, error) {
+    list := make([]Task, 0)
+    err := Db.Where("status = ? AND level = ?", Enabled, TaskLevelParent).
+            Find(&list)
 
-    return list, err
+    if err != nil {
+        return list, err
+    }
+
+    return task.setHostsForTasks(list)
 }
 
 // 获取某个主机下的所有激活任务
-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)
+func (task *Task) ActiveListByHostId(hostId int16) ([]Task, error) {
+    taskHostModel := new(TaskHost)
+    taskIds, err := taskHostModel.GetTaskIdsByHostId(hostId)
+    if err != nil {
+        return nil, err
+    }
+    list := make([]Task, 0)
+    err = Db.Where("status = ?  AND level = ?", Enabled, TaskLevelParent).
+        In("id", taskIds...).
+        Find(&list)
+    if err != nil {
+        return list, err
+    }
 
-    return list, err
+    return task.setHostsForTasks(list)
 }
 
-// 判断主机id是否有引用
-func (task *Task) HostIdExist(hostId int16) (bool, error) {
-    count, err := Db.Where("host_id = ?", hostId).Count(task);
+func (task *Task) setHostsForTasks(tasks []Task) ([]Task, error) {
+    taskHostModel := new(TaskHost)
+    var err error
+    for i, value := range tasks {
+        taskHostDetails, err := taskHostModel.GetHostIdsByTaskId(value.Id)
+        if err != nil {
+            return nil, err
+        }
+        tasks[i].Hosts = taskHostDetails
+    }
 
-    return count > 0, err
+    return tasks, err
 }
 
 // 判断任务名称是否存在
@@ -139,29 +173,63 @@ func (task *Task) GetStatus(id int) (Status, error) {
     return task.Status, nil
 }
 
-func(task *Task) Detail(id int) (TaskHost, error)  {
-    taskHost := TaskHost{}
-    fields := "t.*, host.alias,host.name,host.port"
-    _, err := Db.Alias("t").Join("LEFT", hostTableName(), "t.host_id=host.id").Where("t.id=?", id).Cols(fields).Get(&taskHost)
+func(task *Task) Detail(id int) (Task, error)  {
+    t := Task{}
+    _, err := Db.Where("id=?", id).Get(&t)
 
-    return taskHost, err
+    if err != nil {
+        return t, err
+    }
+
+    taskHostModel := new(TaskHost)
+    t.Hosts, err = taskHostModel.GetHostIdsByTaskId(id)
+
+    return t, err
 }
 
-func (task *Task) List(params CommonMap) ([]TaskHost, error) {
+func (task *Task) List(params CommonMap) ([]Task, error) {
     task.parsePageAndPageSize(params)
-    list := make([]TaskHost, 0)
-    fields := "t.*, host.alias,host.name"
-    session := Db.Alias("t").Join("LEFT", hostTableName(), "t.host_id=host.id")
+    list := make([]Task, 0)
+    session := Db.Alias("t").Join("LEFT", taskHostTableName(), "t.id = th.task_id")
     task.parseWhere(session, params)
-    err := session.Cols(fields).Desc("t.id").Limit(task.PageSize, task.pageLimitOffset()).Find(&list)
+    err := session.GroupBy("t.id").Desc("t.id").Cols("t.*").Limit(task.PageSize, task.pageLimitOffset()).Find(&list)
 
-    return list, err
+    if err != nil {
+        return nil, err
+    }
+
+    return task.setHostsForTasks(list)
+}
+
+// 获取依赖任务列表
+func (task *Task) GetDependencyTaskList(ids string) ([]Task, error) {
+    list := make([]Task, 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.*"
+    err := Db.Alias("t").
+            Where("t.level = ?", TaskLevelChild).
+            In("t.id", taskIds).
+            Cols(fields).
+            Find(&list)
+
+    if err != nil {
+        return list, err
+    }
+
+    return task.setHostsForTasks(list)
 }
 
 func (task *Task) Total(params CommonMap) (int64, error) {
-    session := Db.Alias("t").Join("LEFT", hostTableName(), "t.host_id=host.id")
+    session := Db.Alias("t").Join("LEFT", taskHostTableName(), "t.id = th.task_id")
     task.parseWhere(session, params)
-    return session.Count(task)
+    return session.GroupBy("t.id").Count(task)
 }
 
 // 解析where
@@ -175,7 +243,7 @@ func (task *Task) parseWhere(session *xorm.Session, params CommonMap)  {
     }
     hostId, ok := params["HostId"]
     if ok && hostId.(int) > 0 {
-        session.And("host_id = ?", hostId)
+        session.And("th.host_id = ?", hostId)
     }
     name, ok := params["Name"]
     if ok && name.(string) != "" {
@@ -191,6 +259,3 @@ func (task *Task) parseWhere(session *xorm.Session, params CommonMap)  {
     }
 }
 
-func hostTableName() []string {
-    return []string{TablePrefix + "host", "host"}
-}

models/task_host.go

@@ -0,0 +1,81 @@
+package models
+
+
+type TaskHost struct {
+    Id        int     `xorm:"int pk autoincr"`
+    TaskId    int     `xorm:"int not null index"`
+    HostId    int16   `xorm:"smallint not null index"`
+}
+
+type TaskHostDetail struct {
+    TaskHost `xorm:"extends"`
+    Name string
+    Port int
+    Alias string
+}
+
+func (TaskHostDetail) TableName() string  {
+    return TablePrefix + "task_host"
+}
+
+func hostTableName() []string {
+    return []string{TablePrefix + "host", "h"}
+}
+
+func (th *TaskHost) Remove(taskId int) error {
+    _, err := Db.Where("task_id = ?", taskId).Delete(new(TaskHost))
+
+    return err
+}
+
+func (th *TaskHost) Add(taskId int, hostIds []int) error {
+
+    err := th.Remove(taskId)
+    if err != nil {
+        return err
+    }
+
+    taskHosts := make([]TaskHost, len(hostIds))
+    for i, value := range hostIds {
+        taskHosts[i].TaskId = taskId
+        taskHosts[i].HostId = int16(value)
+    }
+
+    _, err = Db.Insert(&taskHosts)
+
+    return err
+}
+
+func (th *TaskHost) GetHostIdsByTaskId(taskId int) ([]TaskHostDetail, error) {
+    list := make([]TaskHostDetail, 0)
+    fields := "th.id,th.host_id,h.alias,h.name,h.port"
+    err := Db.Alias("th").
+        Join("LEFT", hostTableName(), "th.host_id=h.id").
+        Where("th.task_id = ?", taskId).
+        Cols(fields).
+        Find(&list)
+
+    return list, err
+}
+
+func (th *TaskHost) GetTaskIdsByHostId(hostId int16) ([]interface{}, error)  {
+    list := make([]TaskHost, 0)
+    err := Db.Where("host_id = ?", hostId).Cols("task_id").Find(&list)
+    if err != nil {
+        return nil, err
+    }
+
+    taskIds := make([]interface{}, len(list))
+    for i, value := range list {
+        taskIds[i] = value.TaskId
+    }
+
+    return taskIds, err
+}
+
+// 判断主机id是否有引用
+func (th *TaskHost) HostIdExist(hostId int16) (bool, error) {
+    count, err := Db.Where("host_id = ?", hostId).Count(th);
+
+    return count > 0, err
+}

modules/app/app.go

@@ -4,11 +4,8 @@ import (
     "os"
 
     "github.com/ouqiang/gocron/modules/logger"
-    "runtime"
     "github.com/ouqiang/gocron/modules/utils"
     "gopkg.in/ini.v1"
-    "io/ioutil"
-    "strconv"
 )
 
 var (
@@ -19,12 +16,10 @@ var (
     AppConfig    string // 应用配置文件
     Installed    bool   // 应用是否安装过
     Setting      *ini.Section // 应用配置
-    PidFile      string
 )
 
 
 func InitEnv() {
-    runtime.GOMAXPROCS(runtime.NumCPU())
     logger.InitLogger()
     wd, err := os.Getwd()
     if err != nil {
@@ -35,7 +30,6 @@ func InitEnv() {
     LogDir = AppDir + "/log"
     DataDir = AppDir + "/data"
     AppConfig = ConfDir + "/app.ini"
-    PidFile = LogDir + "/gocron.pid"
     checkDirExists(ConfDir, LogDir, DataDir)
     Installed = IsInstalled()
 }
@@ -50,33 +44,6 @@ func IsInstalled() bool {
     return true
 }
 
-func WritePid()  {
-    pid := os.Getpid()
-    pidStr := strconv.Itoa(pid)
-    err := ioutil.WriteFile(PidFile, []byte(pidStr), 0644)
-    if err != nil {
-        logger.Fatal("写入pid文件失败", err)
-    }
-}
-
-func GetPid() int {
-    bytes, err := ioutil.ReadFile(PidFile)
-    if err != nil {
-        return 0
-    }
-    pidStr := string(bytes)
-    pid, err := strconv.Atoi(pidStr)
-    if err != nil {
-        return 0
-    }
-
-    return pid
-}
-
-func RemovePid()  {
-    os.Remove(PidFile)
-}
-
 // 创建安装锁文件
 func CreateInstallLock() error {
     _, err := os.Create(ConfDir + "/install.lock")

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,13 +30,18 @@ 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)
     }
     s := grpc.NewServer()
     pb.RegisterTaskServer(s, Server{})
-    grpclog.Println("listen address ", addr)
+    grpclog.Println("listen ", addr)
     err = s.Serve(l)
     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)  {
@@ -115,7 +117,7 @@ func Store(ctx *macaron.Context, form HostForm) string  {
             grpcpool.Pool.Release(oldAddr)
         }
 
-        taskModel := new(models.TaskHost)
+        taskModel := new(models.Task)
         tasks, err := taskModel.ActiveListByHostId(id)
         if  err != nil {
             return json.CommonFailure("刷新任务主机信息失败", err)
@@ -133,8 +135,8 @@ func Remove(ctx *macaron.Context) string  {
     if err != nil {
         return json.CommonFailure("参数错误", err)
     }
-    taskModel := new(models.Task)
-    exist,err := taskModel.HostIdExist(int16(id))
+    taskHostModel := new(models.TaskHost)
+    exist,err := taskHostModel.HostIdExist(int16(id))
     if err != nil {
         return json.CommonFailure("操作失败", err)
     }
@@ -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,9 +17,9 @@ 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"
+    "html/template"
 )
 
 // 静态文件目录
@@ -60,17 +60,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 +93,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错误
@@ -143,6 +139,14 @@ func RegisterMiddleware(m *macaron.Macaron) {
         IndentJSON: true,
         // 渲染具有缩进格式的 XML，默认为不缩进
         IndentXML: true,
+        Funcs: []template.FuncMap{map[string]interface{} {
+            "HostFormat": func(index int) bool {
+                return (index + 1) % 3 == 0
+            },
+            "unescape": func(str string) template.HTML {
+                return template.HTML(str)
+            },
+        }},
     }))
     m.Use(session.Sessioner(session.Options{
         Provider:       "file",

routers/task/task.go

@@ -18,14 +18,17 @@ 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)"`
     Multi  int8 `binding:"In(1,2)"`
     RetryTimes int8
-    HostId int16
+    HostId string
     Remark string
     NotifyStatus int8 `binding:"In(1,2,3)"`
     NotifyType int8 `binding:"In(1,2,3)"`
@@ -88,9 +91,22 @@ func Edit(ctx *macaron.Context)  {
         logger.Errorf("编辑任务#获取任务详情失败#任务ID-%d#%s", id, err.Error())
         ctx.Redirect("/task")
     }
+    hostModel := new(models.Host)
+    hostModel.PageSize = -1
+    hosts, err := hostModel.List(models.CommonMap{})
+    if err != nil {
+        logger.Error(err)
+    } else {
+        for i, host := range(hosts) {
+            if inHosts(task.Hosts, host.Id) {
+                hosts[i].Selected = true
+            }
+        }
+    }
+
     ctx.Data["Task"]  = task
+    ctx.Data["Hosts"] = hosts
     ctx.Data["Title"] = "编辑"
-    setHostsToTemplate(ctx)
     ctx.HTML(200, "task/task_form")
 }
 
@@ -99,10 +115,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)
@@ -111,15 +123,10 @@ func Store(ctx *macaron.Context, form TaskForm) string  {
         return json.CommonFailure("任务名称已存在")
     }
 
-    if form.Protocol == models.TaskRPC && form.HostId <= 0 {
+    if form.Protocol == models.TaskRPC && form.HostId == "" {
         return json.CommonFailure("请选择主机名")
     }
 
-    if form.Protocol == models.TaskRPC {
-        taskModel.HostId = form.HostId
-    } else {
-        taskModel.HostId = 0
-    }
     taskModel.Name = form.Name
     taskModel.Protocol = form.Protocol
     taskModel.Command = form.Command
@@ -134,8 +141,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 +161,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 {
         // 任务添加后开始调度执行
@@ -164,8 +195,20 @@ func Store(ctx *macaron.Context, form TaskForm) string  {
         return json.CommonFailure("保存失败", err)
     }
 
+    taskHostModel := new(models.TaskHost)
+    if form.Protocol == models.TaskRPC {
+        hostIdStrList := strings.Split(form.HostId, ",")
+        hostIds := make([]int, len(hostIdStrList))
+        for i, hostIdStr := range hostIdStrList {
+            hostIds[i], _ = strconv.Atoi(hostIdStr)
+        }
+        taskHostModel.Add(id, hostIds)
+    } else {
+        taskHostModel.Remove(id)
+    }
+
     status, err := taskModel.GetStatus(id)
-    if status == models.Enabled {
+    if status == models.Enabled && taskModel.Level == models.TaskLevelParent {
         addTaskToTimer(id)
     }
 
@@ -182,6 +225,9 @@ func Remove(ctx *macaron.Context) string {
         return json.CommonFailure(utils.FailureContent, err)
     }
 
+    taskHostModel := new(models.TaskHost)
+    taskHostModel.Remove(id)
+
     service.Cron.RemoveJob(strconv.Itoa(id))
 
     return json.Success(utils.SuccessContent, nil)
@@ -267,9 +313,20 @@ func parseQueryParams(ctx *macaron.Context) (models.CommonMap) {
 
 func setHostsToTemplate(ctx *macaron.Context)  {
     hostModel := new(models.Host)
+    hostModel.PageSize = -1
     hosts, err := hostModel.List(models.CommonMap{})
     if err != nil {
         logger.Error(err)
     }
     ctx.Data["Hosts"] = hosts
+}
+
+func inHosts(slice []models.TaskHostDetail, element int16) bool {
+    for _, v := range slice {
+        if v.HostId == element {
+            return true
+        }
+    }
+
+    return false
 }

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,9 @@ import (
     "sync"
     rpcClient "github.com/ouqiang/gocron/modules/rpc/client"
     pb "github.com/ouqiang/gocron/modules/rpc/proto"
+    "strings"
+    "text/template"
+    "bytes"
 )
 
 // 定时任务调度管理器
@@ -47,13 +50,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 +69,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 +92,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)
@@ -99,14 +109,18 @@ func (task *Task) Initialize() {
 }
 
 // 批量添加任务
-func (task *Task) BatchAdd(tasks []models.TaskHost)  {
+func (task *Task) BatchAdd(tasks []models.Task)  {
     for _, item := range tasks {
         task.Add(item)
     }
 }
 
 // 添加任务
-func (task *Task) Add(taskModel models.TaskHost) {
+func (task *Task) Add(taskModel models.Task) {
+    if taskModel.Level == models.TaskLevelChild {
+        logger.Errorf("添加任务失败#不允许添加子任务到调度器#任务Id-%d", taskModel.Id);
+        return
+    }
     taskFunc := createJob(taskModel)
     if taskFunc == nil {
         logger.Error("创建任务处理Job失败,不支持的任务协议#", taskModel.Protocol)
@@ -128,12 +142,12 @@ func (task *Task) StopAll()  {
 }
 
 // 直接运行任务
-func (task *Task) Run(taskModel models.TaskHost)  {
+func (task *Task) Run(taskModel models.Task)  {
     go createJob(taskModel)()
 }
 
 type Handler interface {
-    Run(taskModel models.TaskHost) (string, error)
+    Run(taskModel models.Task) (string, error)
 }
 
 
@@ -143,7 +157,7 @@ type HTTPHandler struct{}
 // http任务执行时间不超过300秒
 const HttpExecTimeout = 300
 
-func (h *HTTPHandler) Run(taskModel models.TaskHost) (result string, err error) {
+func (h *HTTPHandler) Run(taskModel models.Task) (result string, err error) {
     if taskModel.Timeout <= 0 || taskModel.Timeout > HttpExecTimeout {
         taskModel.Timeout = HttpExecTimeout
     }
@@ -159,26 +173,54 @@ func (h *HTTPHandler) Run(taskModel models.TaskHost) (result string, err error)
 // RPC调用执行任务
 type RPCHandler struct {}
 
-func (h *RPCHandler) Run(taskModel models.TaskHost) (result string, err error)  {
+func (h *RPCHandler) Run(taskModel models.Task) (result string, err error)  {
     taskRequest := new(pb.TaskRequest)
     taskRequest.Timeout = int32(taskModel.Timeout)
     taskRequest.Command = taskModel.Command
+    var resultChan chan TaskResult = make(chan TaskResult, len(taskModel.Hosts))
+    for _, taskHost := range taskModel.Hosts {
+        go func(th models.TaskHostDetail) {
+            output, err := rpcClient.ExecWithRetry(th.Name, th.Port, taskRequest)
+            var errorMessage string = ""
+            if err != nil {
+                errorMessage = err.Error()
+            }
+            outputMessage := fmt.Sprintf("主机: [%s-%s]\n%s\n%s\n\n",
+                th.Alias, th.Name, errorMessage, output,
+            )
+            resultChan <- TaskResult{Err:err, Result: outputMessage}
+        }(taskHost)
+    }
 
-    return rpcClient.Exec(taskModel.Name, taskModel.Port, taskRequest)
+    var aggregationErr error = nil
+    var aggregationResult string = ""
+    for i := 0; i < len(taskModel.Hosts); i++ {
+        taskResult := <- resultChan
+        aggregationResult += taskResult.Result
+        if taskResult.Err != nil {
+            aggregationErr = taskResult.Err
+        }
+    }
+
+    return aggregationResult, aggregationErr
 }
 
 
 // 创建任务日志
-func createTaskLog(taskModel models.TaskHost, status models.Status) (int64, error) {
+func createTaskLog(taskModel models.Task, status models.Status) (int64, error) {
     taskLogModel := new(models.TaskLog)
     taskLogModel.TaskId = taskModel.Id
-    taskLogModel.Name = taskModel.Task.Name
+    taskLogModel.Name = taskModel.Name
     taskLogModel.Spec = taskModel.Spec
     taskLogModel.Protocol = taskModel.Protocol
     taskLogModel.Command = taskModel.Command
     taskLogModel.Timeout = taskModel.Timeout
     if taskModel.Protocol == models.TaskRPC {
-        taskLogModel.Hostname = taskModel.Alias + "-" + taskModel.Name
+        var aggregationHost string = ""
+        for _, host := range taskModel.Hosts {
+            aggregationHost += fmt.Sprintf("%s-%s<br>", host.Alias, host.Name)
+        }
+        taskLogModel.Hostname = aggregationHost
     }
     taskLogModel.StartTime = time.Now()
     taskLogModel.Status = status
@@ -205,7 +247,7 @@ func updateTaskLog(taskLogId int64, taskResult TaskResult) (int64, error) {
 
 }
 
-func createJob(taskModel models.TaskHost) cron.FuncJob {
+func createJob(taskModel models.Task) cron.FuncJob {
     var handler Handler = createHandler(taskModel)
     if handler == nil {
         return nil
@@ -217,16 +259,16 @@ func createJob(taskModel models.TaskHost) cron.FuncJob {
         if taskLogId <= 0 {
             return
         }
-        logger.Infof("开始执行任务#%s#命令-%s", taskModel.Task.Name, taskModel.Command)
+        logger.Infof("开始执行任务#%s#命令-%s", taskModel.Name, taskModel.Command)
         taskResult := execJob(handler, taskModel)
-        logger.Infof("任务完成#%s#命令-%s", taskModel.Task.Name, taskModel.Command)
+        logger.Infof("任务完成#%s#命令-%s", taskModel.Name, taskModel.Command)
         afterExecJob(taskModel, taskResult, taskLogId)
     }
 
     return taskFunc
 }
 
-func createHandler(taskModel models.TaskHost) Handler  {
+func createHandler(taskModel models.Task) Handler  {
     var handler Handler = nil
     switch taskModel.Protocol {
         case models.TaskHTTP:
@@ -239,7 +281,8 @@ func createHandler(taskModel models.TaskHost) Handler  {
     return handler;
 }
 
-func beforeExecJob(taskModel models.TaskHost) (taskLogId int64)  {
+// 任务前置操作
+func beforeExecJob(taskModel models.Task) (taskLogId int64)  {
     if taskModel.Multi == 0 && runInstance.has(taskModel.Id) {
         createTaskLog(taskModel, models.Cancel)
         return
@@ -258,7 +301,8 @@ func beforeExecJob(taskModel models.TaskHost) (taskLogId int64)  {
     return taskLogId
 }
 
-func afterExecJob(taskModel models.TaskHost, taskResult TaskResult, taskLogId int64)  {
+// 任务执行后置操作
+func afterExecJob(taskModel models.Task, taskResult TaskResult, taskLogId int64)  {
     if taskResult.Err != nil {
         taskResult.Result = taskResult.Err.Error() + "\n" + taskResult.Result
     }
@@ -267,11 +311,78 @@ 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.Task, 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.Command = appendResultToCommand(task.Command, taskResult)
+        task.Spec = fmt.Sprintf("依赖任务(主任务ID-%d)", taskModel.Id)
+        serviceTask.Run(task)
+    }
+}
+
+/**
+ * 添加主任务执行结果到子任务命令中, 占位符{{.Code}} {{.Message}}
+ */
+func appendResultToCommand(command string, taskResult TaskResult) string {
+    var code int8 = 0
+    if taskResult.Err != nil {
+        code = 1
+    }
+    data := map[string]interface{} {
+        "Code": code,
+        "Message": taskResult.Result,
+    }
+    var buf *bytes.Buffer = new(bytes.Buffer)
+    tmpl, err := template.New("command").Parse(command)
+    if err != nil {
+        logger.Errorf("替换子任务命令占位符失败#%s", err.Error())
+        return command
+    }
+    err = tmpl.Execute(buf, data)
+    if err != nil {
+        logger.Errorf("替换子任务命令占位符失败#%s", err.Error())
+        return command
+    }
+
+    return buf.String()
 }
 
 // 发送任务结果通知
-func SendNotification(taskModel models.TaskHost, taskResult TaskResult)  {
+func SendNotification(taskModel models.Task, taskResult TaskResult)  {
     var statusName string
     // 未开启通知
     if taskModel.NotifyStatus == 0 {
@@ -293,7 +404,7 @@ func SendNotification(taskModel models.TaskHost, taskResult TaskResult)  {
     msg := notify.Message{
         "task_type": taskModel.NotifyType,
         "task_receiver_id": taskModel.NotifyReceiverId,
-        "name": taskModel.Task.Name,
+        "name": taskModel.Name,
         "output": taskResult.Result,
         "status": statusName,
         "taskId": taskModel.Id,
@@ -302,7 +413,7 @@ func SendNotification(taskModel models.TaskHost, taskResult TaskResult)  {
 }
 
 // 执行具体任务
-func execJob(handler Handler, taskModel models.TaskHost) TaskResult  {
+func execJob(handler Handler, taskModel models.Task) TaskResult  {
     defer func() {
        if err := recover(); err != nil {
            logger.Error("panic#service/task.go:execJob#", err)

templates/common/header.html

@@ -61,12 +61,11 @@
         <div class="bigcontainer">
             <div class="right menu">
                 <a class="item {{{if or (eq .Controller "task") (eq .Controller "delaytask")}}}active{{{end}}}" href="/task"><i class="tasks icon"></i>任务</a>
-                <a class="item {{{if eq .Controller "host"}}}active{{{end}}}" href="/host"><i class="linux icon"></i>主机</a>
+                <a class="item {{{if eq .Controller "host"}}}active{{{end}}}" href="/host"><i class="linux icon"></i>任务节点</a>
                 <!-- <a class="item {{{if eq .Controller "user"}}}active{{{end}}}" href="/user"><i class="user icon"></i>账户</a> -->
                 {{{if gt .LoginUid 0}}}
                 <a class="item {{{if eq .Controller "manage"}}}active{{{end}}}" href="/manage/slack/edit"><i class="settings icon"></i>管理</a>
                 {{{end}}}
-                <a class="item" href="https://github.com/ouqiang/gocron/wiki" target="_blank"><i class="file text icon"></i>查看文档</a>
             </div>
         </div>
     </div>

templates/host/host_form.html

@@ -29,7 +29,7 @@
                     </div>
                 </div>
                 <div class="field">
-                    <label>主机别名 (方便记忆和引用)</label>
+                    <label>节点名称</label>
                     <div class="ui small  input">
                         <input type="text" name="alias" value="{{{.Host.Alias}}}">
                     </div>

templates/host/index.html

@@ -10,7 +10,7 @@
                     <a href="/host/create">
                         <i class="large add icon"></i>
                         <div class="content">
-                            添加主机
+                            添加节点
                         </div>
                     </a>
                 </h3>
@@ -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/host/menu.html

@@ -2,7 +2,7 @@
     <div class="verticalMenu">
         <div class="ui vertical pointing menu fluid">
             <a class="{{{if eq .URI "/host"}}}active teal{{{end}}}  item" href="/host">
-                <i class="linux icon"></i> 主机列表
+                <i class="linux icon"></i> 节点列表
             </a>
         </div>
     </div>

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

@@ -24,7 +24,7 @@
                 </div>
                 <div class="field">
                     <select name="host_id" id="hostId">
-                        <option value="">选择主机</option>
+                        <option value="">选择节点</option>
                         {{{range $i, $v := .Hosts}}}
                         <option value="{{{.Id}}}" {{{if eq $.Params.HostId .Id }}} selected {{{end}}} >{{{.Alias}}}-{{{.Name}}}</option>
                         {{{end}}}
@@ -54,12 +54,13 @@
             <tr>
                 <th>任务ID</th>
                 <th>任务名称</th>
+                <th>任务类型</th>
                 <th>cron表达式</th>
                 <th>执行方式</th>
                 <th>超时时间</th>
                 <th>重试次数</th>
                 <th>单实例运行</th>
-                <th>主机</th>
+                <th>任务节点</th>
                 <th>状态</th>
                 <th>操作</th>
             </tr>
@@ -68,21 +69,32 @@
                 {{{range $i, $v := .Tasks}}}
                     <tr>
                         <td>{{{.Id}}}</td>
-                        <td>{{{.Task.Name}}}</td>
+                        <td>{{{.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>
+                            {{{range $k, $h := .Hosts}}}
+                                {{{$h.Alias}}}<br>
+                            {{{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 +109,6 @@
     </div>
 </div>
 
-
 <script type="text/javascript">
     $('.ui.checkbox').checkbox();
 

templates/task/log.html

@@ -54,9 +54,9 @@
                 <th>任务ID</th>
                 <th>任务名称</th>
                 <th>cron表达式</th>
-                <th>协议</th>
+                <th>执行方式</th>
                 <th>重试次数</th>
-                <th>主机</th>
+                <th>任务节点</th>
                 <th>执行时长</th>
                 <th>状态</th>
                 <th>执行结果</th>
@@ -70,7 +70,7 @@
                 <td>{{{.Spec}}}</td>
                 <td>{{{if eq .Protocol 1}}} HTTP {{{else if eq .Protocol 2}}} SHELL {{{end}}}</td>
                 <td>{{{.RetryTimes}}}</td>
-                <td>{{{.Hostname}}}</td>
+                <td>{{{unescape .Hostname}}}</td>
                 <td>
                     {{{if and (ne .Status 3) (ne .Status 4)}}}
                     {{{if gt .TotalTime 0}}}{{{.TotalTime}}}秒{{{else}}}1秒{{{end}}}<br>

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.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>
@@ -39,20 +83,23 @@
             <div class="field">
                 <label>执行方式</label>
                 <select name="protocol" id="protocol">
-                    <option value="2" {{{if .Task}}} {{{if eq .Task.Protocol 2}}}selected{{{end}}} {{{end}}} data-match="host_id" data-validate-type="selectProtocol">SHELL</option>
+                    <option value="2" {{{if .Task}}} {{{if eq .Task.Protocol 2}}}selected{{{end}}} {{{end}}}
+                            data-validate-type="selectProtocol">SHELL</option>
                     <option value="1" {{{if .Task}}} {{{if eq .Task.Protocol 1}}}selected{{{end}}} {{{end}}}>HTTP</option>
                 </select>
             </div>
         </div>
-        <div class="three fields" id="hostField">
+        <div class="fields" id="hostField">
             <div class="field">
-                <label>主机</label>
-                <select name="host_id" id="hostId">
-                    <option value="">选择主机</option>
+                <label>选择任务节点</label>
+                <div id="hostId">
                     {{{range $i, $v := .Hosts}}}
-                     <option value="{{{.Id}}}" {{{if $.Task}}}{{{if eq $.Task.HostId .Id }}} selected {{{end}}} {{{end}}}>{{{.Alias}}}-{{{.Name}}}</option>
+                        <label>
+                            <input type="checkbox" value="{{{.Id}}}" {{{if $.Task}}}{{{if $v.Selected}}} checked {{{end}}}{{{end}}}  style="width:25px;height: 25px;">{{{.Alias}}}-{{{.Name}}}
+                            {{{if  (HostFormat $i) }}}<br>{{{end}}}
+                        </label>
                     {{{end}}}
-                 </select> &nbsp; <a class="ui blue button" href="/host/create" target="_blank">添加主机</a>
+                 </div> &nbsp; <br> <a class="ui blue button" href="/host/create" target="_blank">添加节点</a>
 
             </div>
         </div>
@@ -62,19 +109,15 @@
                 <textarea rows="5" name="command" placeholder="请输入系统命令" id="command">{{{.Task.Command}}}</textarea>
             </div>
         </div>
-        <div class="six fields">
-            <div class="field">
-                <label>任务超时时间(秒)</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>
-                <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>任务超时时间(秒, 0-86400)</label>
+                <input type="text"  name="timeout" placeholder="默认0, 不限制" value="{{{if .Task}}} {{{.Task.Timeout}}} {{{else}}}0{{{end}}}">
+            </div>
+            <div class="field">
+                <label>任务失败重试次数 (0-10)</label>
+                <input type="text"  name="retry_times" placeholder="默认0, 不重试" value="{{{if .Task}}} {{{.Task.RetryTimes}}} {{{else}}}0{{{end}}}">
+            </div>
             <div class="field">
                 <label>允许多实例同时运行</label>
                 <select name="multi">
@@ -82,6 +125,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 +151,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 +188,7 @@
 <script type="text/javascript">
     $(function() {
         changeCommandPlaceholder();
+        changeLevel();
         changeProtocol();
         showNotify();
     });
@@ -153,7 +198,9 @@
         changeProtocol();
     });
 
-
+    $('#level').change(function() {
+        changeLevel();
+    });
 
     $('#task-status').change(function() {
         var selected = $(this).val();
@@ -269,6 +316,28 @@
         return receivers.join(",");
     }
 
+    function parseHostId() {
+        var hostIds = [];
+        $('#hostId input:checked').each(function () {
+           hostIds.push($(this).val());
+        });
+
+        return hostIds.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(
@@ -279,6 +348,12 @@
                         return false;
                     }
                     fields.notify_receiver_id = parseNotifyReceiver();
+                    fields.host_id = parseHostId();
+                    if (fields.protocol == 2 && fields.host_id == "") {
+                        swal('错误提示', '请选择任务节点');
+                        return false;
+                    }
+
                     util.post('/task/store', fields, function(code, message) {
                         location.href = "/task"
                     });
@@ -299,19 +374,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",