支持定时任务并发

conf/config.go

@@ -99,6 +99,8 @@ type Config struct {
 
 	CustomCron string `json:"custom_cron" yaml:"custom_cron" mapstructure:"custom_cron"`
 
+	PoolSize int `json:"pool_size" yaml:"pool_size" mapstructure:"pool_size"`
+
 	version string `mapstructure:"version"`
 }
 
@@ -179,6 +181,7 @@ func InitConfig(path string, restart func()) {
 	viper.SetDefault("scheme", "https://johlanse.github.io/study_xxqg/scheme.html?")
 	viper.SetDefault("special_min_score", 10)
 	viper.SetDefault("tg.custom_api", "https://api.telegram.org")
+	viper.SetDefault("pool_size", 1)
 	viper.AutomaticEnv()
 	err := viper.Unmarshal(&config, func(decoderConfig *mapstructure.DecoderConfig) {
 
@@ -195,32 +198,6 @@ func InitConfig(path string, restart func()) {
 			restart()
 		})
 	}
-	//file, err := os.ReadFile(path)
-	//if err != nil {
-	//	log.Warningln("检测到配置文件可能不存在")
-	//	err := os.WriteFile(path, defaultConfig, 0666)
-	//	if err != nil {
-	//		log.Errorln("写入到配置文件出现错误")
-	//		log.Errorln(err.Error())
-	//		return
-	//	}
-	//	log.Infoln("成功写入到配置文件,请重启应用")
-	//	os.Exit(3)
-	//}
-	//err = yaml.Unmarshal(file, &config)
-	//if err != nil {
-	//	log.Errorln(err.Error())
-	//	log.Panicln("配置文件解析失败，请检查配置文件")
-	//}
-	//if config.Scheme == "" {
-	//	config.Scheme = "https://johlanse.github.io/study_xxqg/scheme.html?"
-	//}
-	//if config.SpecialMinScore == 0 {
-	//	config.SpecialMinScore = 10
-	//}
-	//if config.TG.CustomApi == "" {
-	//	config.TG.CustomApi = "https://api.telegram.org"
-	//}
 }
 
 // GetConfig

docs/config.md

@@ -121,5 +121,8 @@ special_min_score: 10
 # 题目搜索的顺序，为true则从2018年最开始搜题，否则从现在最新开始搜题
 reverse_order: false
 
+# 定时任务运行时协程池的大小
+pool_size: 1
+
 
 ```

go.mod

@@ -50,6 +50,7 @@ require (
 	github.com/mattn/go-isatty v0.0.16 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/panjf2000/ants/v2 v2.5.0 // indirect
 	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.5 // indirect
 	github.com/pkg/errors v0.9.1 // indirect

go.sum

@@ -214,6 +214,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
+github.com/panjf2000/ants/v2 v2.5.0 h1:1rWGWSnxCsQBga+nQbA4/iY6VMeNoOIAM0ZWh9u3q2Q=
+github.com/panjf2000/ants/v2 v2.5.0/go.mod h1:cU93usDlihJZ5CfRGNDYsiBYvoilLvBF5Qp/BT2GNRE=
 github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3ve8=
 github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
 github.com/pelletier/go-toml/v2 v2.0.5 h1:ipoSadvV8oGUjnUbMub59IDPPwfxF694nG/jwbMiyQg=

main.go

@@ -11,6 +11,7 @@ import (
 	"os/signal"
 	"path"
 	"strconv"
+	"sync"
 	"syscall"
 	"time"
 
@@ -35,6 +36,7 @@ import (
 var (
 	u          bool
 	i          bool
+	now        bool
 	configPath = ""
 )
 
@@ -55,6 +57,7 @@ func init() {
 
 	flag.BoolVar(&u, "u", false, "更新应用")
 	flag.BoolVar(&i, "init", false, "init the app")
+	flag.BoolVar(&now, "now", false, "run cron now")
 	flag.StringVar(&configPath, "config", "./config/config.yml", "设置配置文件路径")
 	flag.Parse()
 	// 初始化配置文件
@@ -219,8 +222,11 @@ func main() {
 		}
 		c2.Run()
 	}
-
+	inittask()
 	model.SetPush(getPush)
+	if now {
+		do("cron")
+	}
 	if !config.TG.Enable && config.Cron == "" && !config.Wechat.Enable {
 		log.Infoln("已采用普通学习模式")
 		do("normal")
@@ -244,9 +250,6 @@ func do(m string) {
 	getPush := push.GetPush(config)
 	getPush("", "flush", "学习强国助手已上线")
 
-	core := &lib.Core{ShowBrowser: config.ShowBrowser, Push: getPush}
-	defer core.Quit()
-	core.Init()
 	var user *model.User
 	users, _ := model.Query()
 	study := func(core2 *lib.Core, u *model.User) {
@@ -284,21 +287,6 @@ func do(m string) {
 		core2.Push(u.PushId, "flush", message)
 	}
 
-	//c := make(chan *model.User, 1)
-	//
-	//go func() {
-	//	for true {
-	//		u := <-c
-	//		if u.UID == "" {
-	//			break
-	//		} else {
-	//			l := &lib.Core{Push: getPush, ShowBrowser: config.ShowBrowser}
-	//			l.Init()
-	//			study(l, u)
-	//		}
-	//	}
-	//}()
-
 	failUser, _ := model.QueryFailUser()
 	for _, user := range failUser {
 		go func(user2 *model.User) {
@@ -318,6 +306,7 @@ func do(m string) {
 	// 用户小于1时自动登录
 	if len(users) < 1 {
 		log.Infoln("未检测到有效用户信息，将采用登录模式")
+		core := &lib.Core{ShowBrowser: config.ShowBrowser, Push: getPush}
 		u, err := core.L(config.Retry.Times, "")
 		if err != nil {
 			log.Errorln(err.Error())
@@ -325,12 +314,27 @@ func do(m string) {
 		}
 		user = u
 	} else {
+		s := &sync.WaitGroup{}
 		// 如果为定时模式则直接循环所以用户依次运行
 		if m == "cron" {
 			for _, u := range users {
-				study(core, u)
+				//study(core, u)
+				core := &lib.Core{ShowBrowser: config.ShowBrowser, Push: getPush}
+				core.Init()
+				t := &Task{
+					Core: core,
+					User: u,
+					wg:   s,
+				}
+				run(t)
+				s.Add(1)
 			}
+			s.Wait()
 			if len(users) < 1 {
+				core := &lib.Core{ShowBrowser: config.ShowBrowser, Push: getPush}
+
+				core.Init()
+				defer core.Quit()
 				user, err := core.L(config.Retry.Times, "")
 				if err != nil {
 					core.Push(user.PushId, "msg", "登录超时")
@@ -338,6 +342,7 @@ func do(m string) {
 				}
 				study(core, user)
 			}
+			log.Infoln("定时任务执行完成")
 			return
 		}
 
@@ -367,6 +372,7 @@ func do(m string) {
 		}
 
 		if i == 0 {
+			core := &lib.Core{ShowBrowser: config.ShowBrowser, Push: getPush}
 			u, err := core.L(config.Retry.Times, "")
 			if err != nil {
 				log.Errorln(err.Error())
@@ -379,6 +385,10 @@ func do(m string) {
 		}
 	}
 
+	core := &lib.Core{ShowBrowser: config.ShowBrowser, Push: getPush}
+
+	core.Init()
+	defer core.Quit()
 	study(core, user)
 	core.Push(user.PushId, "flush", "")
 }

run.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/panjf2000/ants/v2"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/johlanse/study_xxqg/lib"
+	"github.com/johlanse/study_xxqg/model"
+)
+
+type Task struct {
+	Core *lib.Core
+	User *model.User
+	wg   *sync.WaitGroup
+}
+
+var (
+	pool *ants.PoolWithFunc
+)
+
+func run(task *Task) {
+	pool.Invoke(task)
+}
+
+func inittask() {
+	study := func(core2 *lib.Core, u *model.User) {
+		defer func() {
+			err := recover()
+			if err != nil {
+				log.Errorln("学习过程异常")
+				log.Errorln(err)
+			}
+		}()
+		startTime := time.Now()
+
+		core2.LearnArticle(u)
+
+		core2.LearnVideo(u)
+
+		core2.LearnVideo(u)
+		if config.Model == 2 {
+			core2.RespondDaily(u, "daily")
+		} else if config.Model == 3 {
+			core2.RespondDaily(u, "daily")
+			core2.RespondDaily(u, "weekly")
+			core2.RespondDaily(u, "special")
+		}
+		endTime := time.Now()
+		score, err := lib.GetUserScore(u.ToCookies())
+		if err != nil {
+			log.Errorln("获取成绩失败")
+			log.Debugln(err.Error())
+			return
+		}
+
+		score, _ = lib.GetUserScore(u.ToCookies())
+		message := fmt.Sprintf("%v 学习完成,用时%.1f分钟\n%v", u.Nick, endTime.Sub(startTime).Minutes(), lib.FormatScoreShort(score))
+		core2.Push(u.PushId, "flush", message)
+	}
+
+	pool1, err := ants.NewPoolWithFunc(1, func(i2 interface{}) {
+		task := i2.(*Task)
+		log.Infoln("开始执行" + task.User.Nick)
+		study(task.Core, task.User)
+		defer task.Core.Quit()
+		task.wg.Done()
+	})
+	if err != nil {
+		log.Errorln("创建定时任务协程池失败" + err.Error())
+	}
+	pool = pool1
+}