feat：auto job (#191)

.gitignore

@@ -11,3 +11,4 @@ middleware/demo.go
 config/settings.dev.b.yml
 temp/logs
 config/settings.dev.yml.log
+config/settings.b.dev.yml

apis/sysjob/sysjob.go

@@ -2,12 +2,12 @@ package sysjob
 
 import (
 	"github.com/gin-gonic/gin"
-	"go-admin/global"
+	"go-admin/jobs"
 	"go-admin/models"
-	"go-admin/pkg/cronjob"
 	"go-admin/tools"
 	"go-admin/tools/app"
 	"go-admin/tools/app/msg"
+	"time"
 )
 
 func GetSysJobList(c *gin.Context) {
@@ -82,8 +82,18 @@ func RemoveJob(c *gin.Context) {
 	data.JobId, _ = tools.StringToInt(c.Param("jobId"))
 	result, err := data.Get()
 	tools.HasError(err, "", 500)
-	cronjob.Remove(global.GADMCron, result.EntryId)
+	cn := jobs.Remove(result.EntryId)
-	app.OK(c, nil, msg.DeletedSuccess)
+
+	select {
+	case res := <-cn:
+		if res {
+			_, _ = data.RemoveEntryID(result.EntryId)
+			app.OK(c, nil, msg.DeletedSuccess)
+		}
+	case <-time.After(time.Second * 1):
+		app.OK(c, nil, msg.TimeOut)
+	}
+
 }
 
 func StartJob(c *gin.Context) {
@@ -91,6 +101,14 @@ func StartJob(c *gin.Context) {
 	data.JobId, _ = tools.StringToInt(c.Param("jobId"))
 	result, err := data.Get()
 	tools.HasError(err, "", 500)
-	cronjob.AddJob(result)
+	j := jobs.ExecJob{}
+	j.InvokeTarget = result.InvokeTarget
+	j.CronExpression = result.CronExpression
+	j.JobId = result.JobId
+	j.Name = result.JobName
+	data.EntryId, err = jobs.AddJob(j)
+	tools.HasError(err, "", 500)
+	_, err = data.Update(data.JobId)
+	tools.HasError(err, "", 500)
 	app.OK(c, nil, msg.DeletedSuccess)
 }

cmd/api/server.go

@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/viper"
 	"go-admin/database"
 	"go-admin/global"
+	"go-admin/jobs"
 	mycasbin "go-admin/pkg/casbin"
-	"go-admin/pkg/cronjob"
 	"go-admin/pkg/logger"
 	"go-admin/router"
 	"go-admin/tools"
@@ -76,7 +76,7 @@ func run() error {
 		Handler: r,
 	}
 	go func() {
-		cronjob.JobSetup()
+		jobs.Setup()
 	}()
 
 

config/db.sql

@@ -455,6 +455,9 @@ INSERT INTO sys_user VALUES (1, 'zhangwj', '13818888888', 1, NULL, '', '0', '1@q
 INSERT INTO sys_user VALUES (2, 'zhangwj', '13211111111', 3, NULL, NULL, '0', 'q@q.com', 8, 2, '1', '1', NULL, 0, '2019-11-12 18:28:27', '2020-03-14 20:08:43', NULL, 'zhangwj', '$2a$10$CqMwHahA3cNrNv16CoSxmeD4XMPU.BiKHPEAeaG5oXMavOKrjInXi');
 
 INSERT INTO sys_setting VALUES (1, 'go-admin管理系统', 'https://gitee.com/mydearzwj/image/raw/master/img/go-admin.png', NULL, '2020-08-05 14:27:34', '2020-08-07 10:13:31');
+
+INSERT INTO sys_job VALUES (1, '接口测试', 'DEFAULT', 1, '0/5 * * * * ?', 'http://localhost:8000', 1, 1, 0, '2020-08-03 14:54:03', '2020-08-11 21:48:47', NULL, '', '', 1);
+INSERT INTO sys_job VALUES (2, '函数测试', 'DEFAULT', 2, '0/5 * * * * ?', 'ExamplesNoParam', 1, 1, 0, '2020-08-11 21:41:23', '2020-08-11 21:41:31', NULL, '', '', 2);
 COMMIT;
 
 -- 数据完成 ;

config/pg.sql

@@ -7,4 +7,5 @@ select setval('sys_dict_type_dict_id_seq',12);
 select setval('sys_dict_data_dict_code_seq',32);
 select setval('sys_dept_dept_id_seq',11);
 select setval('sys_config_config_id_seq',4);
+select setval('sys_job_id_seq',3);
 -- 数据完成 ;

global/adm.go

@@ -24,12 +24,11 @@ var (
 var Version string
 
 func init() {
-	Version = "1.1.2"
+	Version = "1.1.3"
 }
 
 var (
 	Logger        *glog.Logger
-	JobLogger      *glog.Logger
+	JobLogger     *glog.Logger
 	RequestLogger *glog.Logger
 )
-

jobs/examples.go

@@ -0,0 +1,16 @@
+package jobs
+
+import (
+	"fmt"
+	"strconv"
+)
+
+func (t *EXEC)ExamplesWithParam(i int , s string) string{
+	fmt.Println("call method PrintInfo i", i, ",s :", s)
+	return s +strconv.Itoa(i)
+}
+
+func (t *EXEC) ExamplesNoParam() string {
+	fmt.Println("\nshow msg input 'call reflect'")
+	return "ShowMsg"
+}

jobs/job.go

@@ -0,0 +1,206 @@
+package jobs
+
+import (
+	"fmt"
+	"github.com/robfig/cron/v3"
+	"go-admin/global"
+	"go-admin/models"
+	"go-admin/pkg"
+	"go-admin/pkg/cronjob"
+	"go-admin/pkg/ws"
+	"reflect"
+	"time"
+)
+
+var timeFormat = "2006-01-02 15:04:05"
+var retryCount = 3
+
+type Job struct {
+	InvokeTarget   string
+	Name           string
+	JobId          int
+	EntryId        int
+	CronExpression string
+}
+
+// 任务类型 http
+type HttpJob struct {
+	Job
+}
+
+type ExecJob struct {
+	Job
+}
+
+func (e ExecJob) Run() {
+	startTime := time.Now()
+
+	if result := callReflect(&EXEC{}, e.InvokeTarget); result != nil {
+		fmt.Printf("callReflectMethod ShowMs %s \n", result[0].String())
+	} else {
+		fmt.Println("callReflectMethod ShowMs didn't run ")
+	}
+
+	// 结束时间
+	endTime := time.Now()
+
+	// 执行时间
+	latencyTime := endTime.Sub(startTime)
+	str := time.Now().Format(timeFormat) + " [INFO] Job " + string(e.EntryId) + "exec success , spend :" + latencyTime.String()
+	ws.SendAll(str)
+	global.JobLogger.Info(time.Now().Format(timeFormat), " [INFO] Job ", e, "exec success , spend :", latencyTime)
+}
+
+//http 任务接口
+func (h HttpJob) Run() {
+
+	startTime := time.Now()
+	var count = 0
+	/* 循环 */
+LOOP:
+	if count < retryCount {
+		/* 跳过迭代 */
+		str, err := pkg.Get(h.InvokeTarget)
+		if err != nil {
+			// 如果失败暂停一段时间重试
+			fmt.Println(time.Now().Format(timeFormat), " [ERROR] mission failed! ", err)
+			fmt.Printf(time.Now().Format(timeFormat)+" [INFO] Retry after the task fails %d seconds! %s \n", time.Duration(count)*time.Second, str)
+			time.Sleep(time.Duration(count) * time.Second)
+			goto LOOP
+		}
+		count = count + 1
+	}
+	// 结束时间
+	endTime := time.Now()
+
+	// 执行时间
+	latencyTime := endTime.Sub(startTime)
+	str := time.Now().Format(timeFormat) + " [INFO] Job " + string(h.EntryId) + "exec success , spend :" + latencyTime.String()
+	ws.SendAll(str)
+	global.JobLogger.Info(time.Now().Format(timeFormat), " [INFO] Job ", h, "exec success , spend :", latencyTime)
+}
+
+// 初始化
+func Setup() {
+
+	fmt.Println(time.Now().Format(timeFormat), " [INFO] Job Starting...")
+
+	global.GADMCron = cronjob.NewWithSeconds()
+
+	sysJob := models.SysJob{}
+	jobList, err := sysJob.GetList()
+	if err != nil {
+		fmt.Println(time.Now().Format(timeFormat), " [ERROR] Job init error", err)
+	}
+	if len(jobList) == 0 {
+		fmt.Println(time.Now().Format(timeFormat), " [INFO] Job total:0")
+	}
+
+	_, err = sysJob.RemoveAllEntryID()
+	if err != nil {
+		fmt.Println(time.Now().Format(timeFormat), " [ERROR] Job remove entry_id error", err)
+	}
+
+	for i := 0; i < len(jobList); i++ {
+		if jobList[i].JobType == 1 {
+			j:=HttpJob{}
+			j.InvokeTarget=jobList[i].InvokeTarget
+			j.CronExpression=jobList[i].CronExpression
+			j.JobId=jobList[i].JobId
+			j.Name=jobList[i].JobName
+
+			sysJob.EntryId, err = AddJob(j)
+		} else if jobList[i].JobType == 2 {
+			j:=ExecJob{}
+			j.InvokeTarget=jobList[i].InvokeTarget
+			j.CronExpression=jobList[i].CronExpression
+			j.JobId=jobList[i].JobId
+			j.Name=jobList[i].JobName
+			sysJob.EntryId, err = AddJob(j)
+		}
+		_, err = sysJob.Update(jobList[i].JobId)
+	}
+
+	// 其中任务
+	global.GADMCron.Start()
+	fmt.Println(time.Now().Format(timeFormat), " [INFO] Job start success.")
+	// 关闭任务
+	defer global.GADMCron.Stop()
+	select {}
+}
+
+// 添加任务 AddJob(invokeTarget string, jobId int, jobName string, cronExpression string)
+func AddJob(job interface{}) (int, error) {
+	switch job.(type) {
+	case HttpJob:
+		op, ok := job.(HttpJob)
+		if ok {
+			return op.addJob()
+		}
+	case ExecJob:
+		op, ok := job.(ExecJob)
+		if ok {
+			return op.addJob()
+		}
+	default:
+		fmt.Println("unknown")
+		return 0, nil
+	}
+	fmt.Println("job error")
+	return 0, nil
+}
+
+func (h HttpJob) addJob() (int, error) {
+	id, err := global.GADMCron.AddJob(h.CronExpression, h)
+	if err != nil {
+		fmt.Println(time.Now().Format(timeFormat), " [ERROR] Job AddJob error", err)
+		return 0, err
+	}
+	EntryId := int(id)
+	return EntryId, nil
+}
+
+func (h ExecJob) addJob() (int, error) {
+	id, err := global.GADMCron.AddJob(h.CronExpression, h)
+	if err != nil {
+		fmt.Println(time.Now().Format(timeFormat), " [ERROR] Job AddJob error", err)
+		return 0, err
+	}
+	EntryId := int(id)
+	return EntryId, nil
+}
+
+// 移除任务
+func Remove(entryID int) chan bool {
+	ch := make(chan bool)
+	go func() {
+		global.GADMCron.Remove(cron.EntryID(entryID))
+		fmt.Println(time.Now().Format(timeFormat), " [INFO] Job Remove success ,info entryID :", entryID)
+		ch <- true
+	}()
+	return ch
+}
+
+// 任务停止
+func Stop() chan bool {
+	ch := make(chan bool)
+	go func() {
+		global.GADMCron.Stop()
+		ch <- true
+	}()
+	return ch
+}
+
+func callReflect(any interface{}, name string, args ...interface{}) []reflect.Value {
+	inputs := make([]reflect.Value, len(args))
+	for i, _ := range args {
+		inputs[i] = reflect.ValueOf(args[i])
+	}
+
+	if v := reflect.ValueOf(any).MethodByName(name); v.String() == "<invalid Value>" {
+		return nil
+	} else {
+		return v.Call(inputs)
+	}
+
+}

jobs/models.go

@@ -0,0 +1,4 @@
+package jobs
+
+type EXEC struct {
+}

models/sysjob.go

@@ -6,17 +6,18 @@ import (
 )
 
 type SysJob struct {
-	JobId          int    `json:"jobId" gorm:"type:int(11);primary_key"`    // 编码
+	JobId          int    `json:"jobId" gorm:"primary_key;AUTO_INCREMENT"` // 编码
-	JobName        string `json:"jobName" gorm:"type:varchar(255);"`        // 名称
+	JobName        string `json:"jobName" gorm:"size:255;"`                // 名称
-	JobGroup       string `json:"jobGroup" gorm:"type:varchar(255);"`       // 任务分组
+	JobGroup       string `json:"jobGroup" gorm:"size:255;"`               // 任务分组
-	CronExpression string `json:"cronExpression" gorm:"type:varchar(255);"` // cron表达式
+	JobType        int    `json:"jobType" gorm:"size:1;"`                  // 任务类型
-	InvokeTarget   string `json:"invokeTarget" gorm:"type:varchar(255);"`   // 调用目标
+	CronExpression string `json:"cronExpression" gorm:"size:255;"`         // cron表达式
-	MisfirePolicy  int    `json:"misfirePolicy" gorm:"type:varchar(255);"`  // 执行策略
+	InvokeTarget   string `json:"invokeTarget" gorm:"size:255;"`           // 调用目标
-	Concurrent     int    `json:"concurrent" gorm:"type:int(1);"`           // 是否并发
+	MisfirePolicy  int    `json:"misfirePolicy" gorm:"size:255;"`          // 执行策略
-	Status         int    `json:"status" gorm:"type:int(1);"`               // 状态
+	Concurrent     int    `json:"concurrent" gorm:"size:1;"`               // 是否并发
-	EntryId        int    `json:"entry_id" gorm:"type:int(11);"`            // job启动时返回的id
+	Status         int    `json:"status" gorm:"size:1;"`                   // 状态
-	CreateBy       string `json:"createBy" gorm:"type:varchar(128);"`       //
+	EntryId        int    `json:"entry_id" gorm:"size:11;"`                // job启动时返回的id
-	UpdateBy       string `json:"updateBy" gorm:"type:varchar(128);"`       //
+	CreateBy       string `json:"createBy" gorm:"size:128;"`               //
+	UpdateBy       string `json:"updateBy" gorm:"size:128;"`               //
 	DataScope      string `json:"dataScope" gorm:"-"`
 	BaseModel
 }

pkg/cronjob/gadmjob.go

@@ -1,134 +1,12 @@
 package cronjob
 
 import (
-	"fmt"
 	"github.com/robfig/cron/v3"
-	"go-admin/global"
-	"go-admin/models"
-	"go-admin/pkg"
-	"go-admin/pkg/ws"
-	"time"
 )
 
-var timeFormat = "2006-01-02 15:04:05"
-var retryCount = 3
-
-// 初始化
-func JobSetup() {
-
-	fmt.Println(time.Now().Format(timeFormat), " [INFO] Job Starting...")
-
-	global.GADMCron = newWithSeconds()
-
-	sysjob := models.SysJob{}
-	joblist, err := sysjob.GetList()
-	if err != nil {
-		fmt.Println(time.Now().Format(timeFormat), " [ERROR] Job init error", err)
-	}
-	if len(joblist) == 0 {
-		fmt.Println(time.Now().Format(timeFormat), " [INFO] Job total:0")
-	}
-
-	_, err = sysjob.RemoveAllEntryID()
-	if err != nil {
-		fmt.Println(time.Now().Format(timeFormat), " [ERROR] Job remove entry_id error", err)
-	}
-
-	for i := 0; i < len(joblist); i++ {
-		AddJob(joblist[i])
-
-	}
-
-	// 其中任务
-	global.GADMCron.Start()
-	fmt.Println(time.Now().Format(timeFormat), " [INFO] Job start success.")
-	// 关闭任务
-	defer global.GADMCron.Stop()
-	select {}
-}
-
-// 添加任务
-func AddJob(job models.SysJob) error {
-	h := HttpJob{Url: job.InvokeTarget, JobId: job.JobId, Name: job.JobName}
-	// 添加定时任务
-	id, err := global.GADMCron.AddJob(job.CronExpression, h)
-
-	if err != nil {
-		fmt.Println(time.Now().Format(timeFormat), " [ERROR] Job AddJob error", err)
-		return err
-	}
-	job.EntryId = int(id)
-	h.EntryId = job.EntryId
-	_, err = job.Update(job.JobId)
-	fmt.Println(time.Now().Format(timeFormat), " [INFO] Job AddJob success ,info:", job)
-	return err
-}
-
-// 任务类型 http
-type HttpJob struct {
-	Url     string
-	Name    string
-	JobId   int
-	EntryId int
-}
-
-//http 任务接口
-func (h HttpJob) Run() {
-
-	startTime := time.Now()
-	var count = 0
-	/* 循环 */
-LOOP:
-	if count < retryCount {
-		/* 跳过迭代 */
-		str, err := pkg.Get(h.Url)
-		if err != nil {
-			// 如果失败暂停一段时间重试
-			fmt.Println(time.Now().Format(timeFormat), " [ERROR] mission failed! ", err)
-			fmt.Printf(time.Now().Format(timeFormat)+" [INFO] Retry after the task fails %d seconds! %s \n", time.Duration(count)*time.Second, str)
-			time.Sleep(time.Duration(count) * time.Second)
-			goto LOOP
-		}
-		count = count + 1
-		//fmt.Printf("a的值为 : %s \n", str)
-	}
-
-	fmt.Println(h.Url)
-	// 结束时间
-	endTime := time.Now()
-
-	// 执行时间
-	latencyTime := endTime.Sub(startTime)
-	str := time.Now().Format(timeFormat) + " [INFO] Job " + string(h.EntryId) + "exec success , spend :" + latencyTime.String()
-	ws.SendAll(str)
-	global.JobLogger.Info(time.Now().Format(timeFormat), " [INFO] Job ", h, "exec success , spend :", latencyTime)
-}
-
-// 调度停止
-func Stop(cron *cron.Cron) chan bool {
-	ch := make(chan bool)
-	go func() {
-		cron.Stop()
-		ch <- true
-	}()
-	return ch
-}
-
-//移除Job
-func Remove(e *cron.Cron, entryID int) {
-	ch := make(chan bool)
-	job := models.SysJob{}
-	go func() {
-
-		e.Remove(cron.EntryID(entryID))
-		_, _ = job.RemoveEntryID(entryID)
-		fmt.Println(time.Now().Format(timeFormat), " [INFO] Job Remove success ,info entryID :", entryID)
-		ch <- true
-	}()
-}
 
 // newWithSeconds returns a Cron with the seconds field enabled.
-func newWithSeconds() *cron.Cron {
+func NewWithSeconds() *cron.Cron {
 	secondParser := cron.NewParser(cron.Second | cron.Minute |
 		cron.Hour | cron.Dom | cron.Month | cron.DowOptional | cron.Descriptor)
 	return cron.New(cron.WithParser(secondParser), cron.WithChain())

tools/app/msg/message.go

@@ -7,4 +7,5 @@ var (
 	DeletedFail    = "删除失败！"
 	GetSuccess     = "查询成功！"
 	NotFound       = "未找到相关内容或者数据为空！"
+	TimeOut        = "操作超时！"
 )