diff --git a/cmd/web.go b/cmd/web.go index 07852c1..111fbf8 100644 --- a/cmd/web.go +++ b/cmd/web.go @@ -18,6 +18,7 @@ import ( "path/filepath" "os/exec" "github.com/ouqiang/gocron/modules/utils" + "github.com/ouqiang/gocron/modules/rpc/grpcpool" ) // web服务器默认端口 @@ -220,6 +221,9 @@ func shutdown() { serviceDelayTask := new(service.DelayTask) serviceDelayTask.Stop() } + // 释放gRPC连接池 + grpcpool.Pool.ReleaseAll() + taskNumInRunning := service.TaskNum.Num() logger.Infof("正在运行的任务有%d个", taskNumInRunning) if taskNumInRunning > 0 { diff --git a/modules/rpc/grpcpool/grpc_pool.go b/modules/rpc/grpcpool/grpc_pool.go index 9084e86..a550278 100644 --- a/modules/rpc/grpcpool/grpc_pool.go +++ b/modules/rpc/grpcpool/grpc_pool.go @@ -74,6 +74,15 @@ func (p *GRPCPool) Release(addr string) { delete(p.conns, addr) } +// 释放所有连接池 +func (p *GRPCPool) ReleaseAll() { + p.Lock() + defer p.Unlock() + for _, pool := range(p.conns) { + pool.Release() + } +} + // 初始化底层连接池 func (p *GRPCPool) newCommonPool(addr string) (error) { p.Lock() diff --git a/routers/task/task.go b/routers/task/task.go index d0c9ee5..2f32b75 100644 --- a/routers/task/task.go +++ b/routers/task/task.go @@ -28,7 +28,7 @@ type TaskForm struct { HostId int16 Remark string NotifyStatus int8 `binding:"In(1,2,3)"` - NotifyType int8 `binding:"In(1,2)"` + NotifyType int8 `binding:"In(1,2,3)"` NotifyReceiverId string } @@ -245,7 +245,7 @@ func addTaskToTimer(id int) { } taskService := service.Task{} - taskService.Add(&task) + taskService.Add(task) } // 解析查询参数 diff --git a/service/task.go b/service/task.go index 20fa36e..e11298f 100644 --- a/service/task.go +++ b/service/task.go @@ -101,12 +101,12 @@ func (task *Task) Initialize() { // 批量添加任务 func (task *Task) BatchAdd(tasks []models.TaskHost) { for _, item := range tasks { - task.Add(&item) + task.Add(item) } } // 添加任务 -func (task *Task) Add(taskModel *models.TaskHost) { +func (task *Task) Add(taskModel models.TaskHost) { taskFunc := createJob(taskModel) if taskFunc == nil { logger.Error("创建任务处理Job失败,不支持的任务协议#", taskModel.Protocol) @@ -129,11 +129,11 @@ func (task *Task) StopAll() { // 直接运行任务 func (task *Task) Run(taskModel models.TaskHost) { - go createJob(&taskModel)() + go createJob(taskModel)() } type Handler interface { - Run(taskModel *models.TaskHost) (string, error) + Run(taskModel models.TaskHost) (string, error) } @@ -143,7 +143,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.TaskHost) (result string, err error) { if taskModel.Timeout <= 0 || taskModel.Timeout > HttpExecTimeout { taskModel.Timeout = HttpExecTimeout } @@ -159,7 +159,7 @@ 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.TaskHost) (result string, err error) { taskRequest := new(pb.TaskRequest) taskRequest.Timeout = int32(taskModel.Timeout) taskRequest.Command = taskModel.Command @@ -169,7 +169,7 @@ func (h *RPCHandler) Run(taskModel *models.TaskHost) (result string, err error) // 创建任务日志 -func createTaskLog(taskModel *models.TaskHost, status models.Status) (int64, error) { +func createTaskLog(taskModel models.TaskHost, status models.Status) (int64, error) { taskLogModel := new(models.TaskLog) taskLogModel.TaskId = taskModel.Id taskLogModel.Name = taskModel.Task.Name @@ -205,7 +205,7 @@ func updateTaskLog(taskLogId int64, taskResult TaskResult) (int64, error) { } -func createJob(taskModel *models.TaskHost) cron.FuncJob { +func createJob(taskModel models.TaskHost) cron.FuncJob { var handler Handler = createHandler(taskModel) if handler == nil { return nil @@ -226,7 +226,7 @@ func createJob(taskModel *models.TaskHost) cron.FuncJob { return taskFunc } -func createHandler(taskModel *models.TaskHost) Handler { +func createHandler(taskModel models.TaskHost) Handler { var handler Handler = nil switch taskModel.Protocol { case models.TaskHTTP: @@ -235,10 +235,11 @@ func createHandler(taskModel *models.TaskHost) Handler { handler = new(RPCHandler) } + return handler; } -func beforeExecJob(taskModel *models.TaskHost) (taskLogId int64) { +func beforeExecJob(taskModel models.TaskHost) (taskLogId int64) { if taskModel.Multi == 0 && runInstance.has(taskModel.Id) { createTaskLog(taskModel, models.Cancel) return @@ -257,7 +258,7 @@ func beforeExecJob(taskModel *models.TaskHost) (taskLogId int64) { return taskLogId } -func afterExecJob(taskModel *models.TaskHost, taskResult TaskResult, taskLogId int64) { +func afterExecJob(taskModel models.TaskHost, taskResult TaskResult, taskLogId int64) { if taskResult.Err != nil { taskResult.Result = taskResult.Err.Error() + "\n" + taskResult.Result } @@ -270,7 +271,7 @@ func afterExecJob(taskModel *models.TaskHost, taskResult TaskResult, taskLogId i } // 发送任务结果通知 -func SendNotification(taskModel *models.TaskHost, taskResult TaskResult) { +func SendNotification(taskModel models.TaskHost, taskResult TaskResult) { var statusName string // 未开启通知 if taskModel.NotifyStatus == 0 { @@ -301,7 +302,7 @@ func SendNotification(taskModel *models.TaskHost, taskResult TaskResult) { } // 执行具体任务 -func execJob(handler Handler, taskModel *models.TaskHost) TaskResult { +func execJob(handler Handler, taskModel models.TaskHost) TaskResult { defer func() { if err := recover(); err != nil { logger.Error("panic#service/task.go:execJob#", err)