等待任务结束从轮询改为WaitGroup

cmd/web.go

@@ -4,7 +4,6 @@ import (
 	"github.com/ouqiang/gocron/models"
 	"github.com/ouqiang/gocron/modules/app"
 	"github.com/ouqiang/gocron/modules/logger"
-	"github.com/ouqiang/gocron/modules/rpc/grpcpool"
 	"github.com/ouqiang/gocron/modules/setting"
 	"github.com/ouqiang/gocron/routers"
 	"github.com/ouqiang/gocron/service"
@@ -13,7 +12,6 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
-	"time"
 )
 
 // web服务器默认端口
@@ -152,23 +150,7 @@ func shutdown() {
 	serviceTask := new(service.Task)
 	// 停止所有任务调度
 	logger.Info("停止定时任务调度")
-	serviceTask.StopAll()
+	serviceTask.Stop()
-
-	taskNumInRunning := service.TaskNum.Num()
-	logger.Infof("正在运行的任务有%d个", taskNumInRunning)
-	if taskNumInRunning > 0 {
-		logger.Info("等待所有任务执行完成后退出")
-	}
-	for {
-		if taskNumInRunning <= 0 {
-			break
-		}
-		time.Sleep(3 * time.Second)
-		taskNumInRunning = service.TaskNum.Num()
-	}
-
-	// 释放gRPC连接池
-	grpcpool.Pool.ReleaseAll()
 }
 
 // 判断应用是否需要升级, 当存在版本号文件且版本小于app.VersionId时升级

modules/app/app.go

@@ -9,6 +9,7 @@ import (
 	"io/ioutil"
 	"strconv"
 	"strings"
+	"fmt"
 )
 
 var (
@@ -35,7 +36,7 @@ func InitEnv(versionString string) {
 	DataDir = AppDir + "/data"
 	AppConfig = ConfDir + "/app.ini"
 	VersionFile = ConfDir + "/.version"
-	checkDirExists(ConfDir, LogDir, DataDir)
+	createDirIfNeed(ConfDir, LogDir, DataDir)
 	Installed = IsInstalled()
 	VersionId = ToNumberVersion(versionString)
 }
@@ -107,10 +108,13 @@ func ToNumberVersion(versionString string) int {
 }
 
 // 检测目录是否存在
-func checkDirExists(path ...string) {
+func createDirIfNeed(path ...string) {
 	for _, value := range path {
 		if !utils.FileExist(value) {
-			logger.Fatal(value + "目录不存在或无权限访问")
+			err := os.Mkdir(value, 0755)
+			if err != nil {
+				logger.Fatal(fmt.Sprintf("创建目录失败:%s", err.Error()))
+			}
 		}
 	}
 }

service/task.go

@@ -23,61 +23,51 @@ var Cron *cron.Cron
 var runInstance Instance
 
 // 任务计数-正在运行中的任务
-var TaskNum TaskCount
+var taskCount TaskCount
 
 // 任务计数
 type TaskCount struct {
-	num int
+	wg sync.WaitGroup
-	sync.RWMutex
+	exit chan bool
 }
 
-func (c *TaskCount) Add() {
+func (tc *TaskCount) Add()  {
-	c.Lock()
+	tc.wg.Add(1)
-	defer c.Unlock()
-	c.num += 1
 }
 
-func (c *TaskCount) Done() {
+func (tc *TaskCount) Done()  {
-	c.Lock()
+	tc.wg.Done()
-	defer c.Unlock()
-	c.num -= 1
 }
 
-func (c *TaskCount) Num() int {
+func (tc *TaskCount) Exit()  {
-	c.RLock()
+	tc.wg.Done()
-	defer c.RUnlock()
+	<-tc.exit
+}
 
-	return c.num
+func (tc *TaskCount) Wait()  {
+	tc.Add()
+	tc.wg.Wait()
+	close(tc.exit)
 }
 
 // 任务ID作为Key
 type Instance struct {
-	Status map[int]bool
+	m sync.Map
-	sync.RWMutex
 }
 
 // 是否有任务处于运行中
 func (i *Instance) has(key int) bool {
-	i.RLock()
+	_, ok := i.m.Load(key)
-	defer i.RUnlock()
-	running, ok := i.Status[key]
-	if ok && running {
-		return true
-	}
 
-	return false
+	return ok
 }
 
 func (i *Instance) add(key int) {
-	i.Lock()
+	i.m.Store(key, true)
-	defer i.Unlock()
-	i.Status[key] = true
 }
 
 func (i *Instance) done(key int) {
-	i.Lock()
+	i.m.Delete(key)
-	defer i.Unlock()
-	delete(i.Status, key)
 }
 
 type Task struct{}
@@ -92,8 +82,9 @@ type TaskResult struct {
 func (task *Task) Initialize() {
 	Cron = cron.New()
 	Cron.Start()
-	runInstance = Instance{make(map[int]bool), sync.RWMutex{}}
+	runInstance = Instance{}
-	TaskNum = TaskCount{0, sync.RWMutex{}}
+	taskCount = TaskCount{sync.WaitGroup{}, make(chan bool)}
+	go taskCount.Wait()
 
 	taskModel := new(models.Task)
 	taskList, err := taskModel.ActiveList()
@@ -137,8 +128,9 @@ func (task *Task) Add(taskModel models.Task) {
 }
 
 // 停止所有任务
-func (task *Task) StopAll() {
+func (task *Task) Stop() {
 	Cron.Stop()
+	taskCount.Exit()
 }
 
 // 直接运行任务
@@ -251,8 +243,8 @@ func createJob(taskModel models.Task) cron.FuncJob {
 		return nil
 	}
 	taskFunc := func() {
-		TaskNum.Add()
+		taskCount.Add()
-		defer TaskNum.Done()
+		defer taskCount.Done()
 		taskLogId := beforeExecJob(taskModel)
 		if taskLogId <= 0 {
 			return