feat: background task pool support

Author: ka1i

internal/app/cli/cli.go

@@ -12,6 +12,7 @@ import (
 	"github.com/ka1i/cli/internal/pkg/datasource"
 	"github.com/ka1i/cli/internal/pkg/handlers"
 	"github.com/ka1i/cli/internal/pkg/system/prepare"
+	"github.com/ka1i/cli/internal/pkg/task"
 	"github.com/ka1i/cli/pkg/logger"
 	"github.com/ka1i/cli/pkg/utils"
 )
@@ -44,6 +45,9 @@ func Execute() error {
 	// system initialize
 	prepare.Configure()
 
+	// service background task
+	go task.New.Start()
+
 	// configure web server
 	engine := gin.New()
 	handlers.SetHandlers(engine)

internal/pkg/config/model.go

@@ -9,6 +9,8 @@ type Options struct {
 
 	Mysql `yaml:"mysql"` // 数据库配置
 	Redis `yaml:"redis"` // 数据库配置
+
+	Task `yaml:"task"` // 后台任务配置
 }
 
 // app env config
@@ -43,3 +45,7 @@ type Redis struct {
 	Passwd     string   `yaml:"passwd"`     // 密码
 	MasterName string   `yaml:"mastername"` // 选项
 }
+
+type Task struct {
+	Interval time.Duration `yaml:"interval"` // 间隔
+}

internal/pkg/task/init.go

@@ -0,0 +1,13 @@
+package task
+
+import (
+	"runtime"
+)
+
+var (
+	waitCount int = 0
+
+	waitLatestTaskDone = make(chan bool)
+
+	goroutineMaxProcess = make(chan bool, runtime.NumCPU()*100)
+)

internal/pkg/task/task.go

@@ -0,0 +1,68 @@
+package task
+
+import (
+	"sync"
+	"time"
+
+	"github.com/ka1i/cli/internal/pkg/config"
+	"github.com/ka1i/cli/pkg/logger"
+	"go.uber.org/zap"
+)
+
+var New = getTask()
+
+func getTask() *taskbBase {
+	return &taskbBase{}
+}
+
+type taskbBase struct {
+	surveillance time.Duration
+}
+
+func (t *taskbBase) init() {
+	time.Sleep(time.Second * 1)
+	interval := config.Cfg.Get().Interval
+	if interval == 0 {
+		logger.Error("Please check background task interval, Use defaults interval: 1min")
+		interval = time.Second * 60
+	}
+	t.surveillance = interval / 2
+
+	logger.Info("background task", zap.String("interval", interval.String()), zap.String("surveillance", t.surveillance.String()))
+}
+
+func (t *taskbBase) Start() {
+
+	t.init()
+
+	var wg sync.WaitGroup
+	surveillanceTicker := time.NewTicker(t.surveillance)
+
+	var waitLatestTask bool = false
+	for {
+		select {
+		case <-surveillanceTicker.C:
+			// surveillance task
+			if waitLatestTask { // check result wait
+				waitCount++
+				if waitCount == 1 {
+					go func() {
+						wg.Wait()
+						waitLatestTaskDone <- true
+					}()
+				}
+			} else { // start surveillance group by
+				wg.Add(1)
+				go surveillance(&wg)
+				waitLatestTask = true
+			}
+		case <-waitLatestTaskDone:
+			if waitCount > 1 {
+				logger.Warnf("注意，请调整任务间隔，建议调整间隔最小为: %v~%v\n", t.surveillance*time.Duration(waitCount-1), t.surveillance*time.Duration(waitCount+1))
+			}
+			// analyse surveillance task result
+			waitCount = 0
+			waitLatestTask = false
+		}
+	}
+}

internal/pkg/task/worker.go

@@ -0,0 +1,19 @@
+package task
+
+import (
+	"sync"
+
+	"github.com/ka1i/cli/pkg/logger"
+)
+
+func surveillance(wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	wg.Add(1)
+	goroutineMaxProcess <- true
+	go func() {
+		defer wg.Done()
+		defer func() { <-goroutineMaxProcess }()
+		logger.Info("new task running...")
+	}()
+}