237 lines
7.2 KiB
Go
237 lines
7.2 KiB
Go
package logic
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/net/gclient"
|
||
"github.com/gogf/gf/v2/os/gcron"
|
||
|
||
"easygoadmin/internal/dao"
|
||
"easygoadmin/utility/cache"
|
||
)
|
||
|
||
// ============================================================================
|
||
// 定时任务调度器：基于 gcron 的常驻调度，启动时加载 status=1 的任务，
|
||
// 按 cron_expression 触发 url+params（HTTP 调用本服务接口），结果写任务日志表。
|
||
//
|
||
// 任务键：job:{id}（gcron 命名条目），暂停=Remove、恢复=重新 Add。
|
||
// ============================================================================
|
||
|
||
var (
|
||
daemonOnce sync.Once // 保证调度器仅初始化一次
|
||
daemonStart time.Time // 调度器启动时间
|
||
)
|
||
|
||
// jobLockTTLSeconds 任务执行锁 TTL（秒）
|
||
// 兜底上限：执行实例崩溃未主动释放时锁自动过期；正常路径执行结束即释放
|
||
const jobLockTTLSeconds = 300
|
||
|
||
// jobTryLock 抢占任务执行锁
|
||
//
|
||
// 键 job:lock:{id} 经 utility/cache（Redis 优先、内存降级）SET NX 实现，
|
||
// 多实例部署共享同一把锁，防止同一任务被并发执行；
|
||
// 抢占成功返回释放函数，失败返回 nil（调用方放弃本次执行）
|
||
func jobTryLock(jobId int64) func() {
|
||
key := fmt.Sprintf("job:lock:%d", jobId)
|
||
ctx := context.Background()
|
||
if !cache.SetNX(ctx, key, "1", jobLockTTLSeconds) {
|
||
return nil
|
||
}
|
||
return func() { _ = cache.Del(ctx, key) }
|
||
}
|
||
|
||
// jobKey 生成 gcron 条目键
|
||
func jobKey(jobId int64) string {
|
||
return fmt.Sprintf("job:%d", jobId)
|
||
}
|
||
|
||
// DaemonInit 启动时加载运行中的任务
|
||
// 幂等，多次调用仅首次生效
|
||
func DaemonInit(ctx context.Context) {
|
||
daemonOnce.Do(func() {
|
||
daemonStart = time.Now()
|
||
rows, err := dao.Job.Ctx(ctx).Where("status", 1).All()
|
||
if err != nil {
|
||
g.Log().Warningf(ctx, "定时任务加载失败: %v", err)
|
||
return
|
||
}
|
||
g.Log().Debugf(ctx, "定时任务待挂载: %d 条", len(rows))
|
||
loaded := 0
|
||
for _, r := range rows {
|
||
if err := addJobToScheduler(r["id"].Int64(), r["cron_expression"].String(),
|
||
r["execute_policy"].Int(), r["is_sync"].Int()); err != nil {
|
||
g.Log().Warningf(ctx, "任务挂载失败 job=%d cron=%q: %v",
|
||
r["id"].Int64(), r["cron_expression"].String(), err)
|
||
continue
|
||
}
|
||
loaded++
|
||
}
|
||
g.Log().Infof(ctx, "定时任务调度器已启动，加载 %d 个任务", loaded)
|
||
})
|
||
}
|
||
|
||
// addJobToScheduler 挂载任务到调度器
|
||
//
|
||
// 执行策略（execute_policy）：
|
||
// - 1 立即执行：允许并发，tick 即启动新执行（gcron.Add）
|
||
// - 2 执行一次（默认）：单例调度，上次未结束则跳过本次（gcron.AddSingleton）
|
||
// - 3 放弃执行：并发 tick 照常触发，但抢不到执行锁则放弃本次
|
||
//
|
||
// 执行方式（is_sync）：1 同步在调度协程内执行；0 异步 go 协程执行。
|
||
// 策略 2/3 均叠加执行锁（多实例共享），策略 1 按并发语义不加锁。
|
||
func addJobToScheduler(jobId int64, cronExpr string, executePolicy int, isSync int) error {
|
||
if cronExpr == "" {
|
||
return fmt.Errorf("cron表达式为空")
|
||
}
|
||
handler := func(c context.Context) {
|
||
run := func(ctx context.Context) {
|
||
if executePolicy != 1 {
|
||
unlock := jobTryLock(jobId)
|
||
if unlock == nil {
|
||
g.Log().Debugf(ctx, "任务 %d 存在执行中的实例，本次放弃", jobId)
|
||
return
|
||
}
|
||
defer unlock()
|
||
}
|
||
executeJob(ctx, jobId, false)
|
||
}
|
||
if isSync == 1 {
|
||
run(c)
|
||
} else {
|
||
go run(context.Background())
|
||
}
|
||
}
|
||
var err error
|
||
if executePolicy == 1 {
|
||
_, err = gcron.Add(context.Background(), cronExpr, handler, jobKey(jobId))
|
||
} else {
|
||
_, err = gcron.AddSingleton(context.Background(), cronExpr, handler, jobKey(jobId))
|
||
}
|
||
return err
|
||
}
|
||
|
||
// removeJobFromScheduler 从调度器摘除任务
|
||
func removeJobFromScheduler(jobId int64) {
|
||
gcron.Remove(jobKey(jobId))
|
||
}
|
||
|
||
// hasJobInScheduler 判断调度器中是否包含指定任务
|
||
func hasJobInScheduler(jobId int64) bool {
|
||
return gcron.Search(jobKey(jobId)) != nil
|
||
}
|
||
|
||
// executeJob 执行任务
|
||
//
|
||
// 流程：读取任务定义 → HTTP 调用目标地址 → 写入执行日志；
|
||
// 内部捕获 panic，避免异常外泄影响调度器。
|
||
// force 为 true 时跳过状态检查，用于手动"立即执行"场景，未发布任务也可执行。
|
||
func executeJob(ctx context.Context, jobId int64, force bool) {
|
||
defer func() {
|
||
if e := recover(); e != nil {
|
||
g.Log().Errorf(ctx, "定时任务 %d 执行 panic: %v", jobId, e)
|
||
}
|
||
}()
|
||
one, err := dao.Job.Ctx(ctx).Where("id", jobId).One()
|
||
if err != nil || len(one) == 0 {
|
||
removeJobFromScheduler(jobId)
|
||
return
|
||
}
|
||
if !force && one["status"].Int() != 1 {
|
||
removeJobFromScheduler(jobId)
|
||
return
|
||
}
|
||
|
||
start := time.Now()
|
||
status := 0
|
||
message := "执行成功"
|
||
exception := ""
|
||
|
||
// 任务执行目标分派：
|
||
// http(s) 地址 → HTTP 调用；其余取值 → 按方法名查找本地处理方法执行
|
||
target := one["url"].String()
|
||
if target == "" {
|
||
status = 1
|
||
message = "执行失败"
|
||
exception = "任务URL为空"
|
||
} else if strings.HasPrefix(target, "http://") || strings.HasPrefix(target, "https://") {
|
||
client := gclient.New()
|
||
client.SetContentType("application/json")
|
||
client.SetTimeout(30 * time.Second)
|
||
body := one["params"].String()
|
||
var response *gclient.Response
|
||
var rerr error
|
||
if body == "" {
|
||
response, rerr = client.Get(ctx, target)
|
||
} else {
|
||
response, rerr = client.Post(ctx, target, body)
|
||
}
|
||
if rerr != nil {
|
||
status = 1
|
||
message = "执行失败"
|
||
exception = truncStr(rerr.Error(), 1000)
|
||
} else {
|
||
respBody := truncStr(response.ReadAllString(), 1000)
|
||
if response.StatusCode != 200 {
|
||
status = 1
|
||
message = "执行失败"
|
||
exception = fmt.Sprintf("HTTP %d: %s", response.StatusCode, respBody)
|
||
} else {
|
||
message = truncStr("执行成功: "+respBody, 500)
|
||
}
|
||
}
|
||
} else {
|
||
handler, ok := jobHandlers[target]
|
||
if !ok {
|
||
status = 1
|
||
message = "执行失败"
|
||
exception = fmt.Sprintf("未找到本地任务处理方法: %s", truncStr(target, 200))
|
||
} else {
|
||
summary, herr := handler(ctx, one["params"].String())
|
||
if herr != nil {
|
||
status = 1
|
||
message = "执行失败"
|
||
exception = truncStr(herr.Error(), 1000)
|
||
} else if summary != "" {
|
||
message = truncStr("执行成功: "+summary, 500)
|
||
}
|
||
}
|
||
}
|
||
end := time.Now()
|
||
|
||
// 执行日志落库（后台 ctx，独立于任务 ctx）
|
||
if _, werr := dao.JobLog.Ctx(ctx).Data(map[string]any{
|
||
"job_id": jobId,
|
||
"job_name": one["job_name"].String(),
|
||
"job_group": one["job_group"].String(),
|
||
"job_trigger": one["job_trigger"].String(),
|
||
"job_message": message,
|
||
"cron_expression": one["cron_expression"].String(),
|
||
"status": status,
|
||
"start_time": start,
|
||
"end_time": end,
|
||
"consume_time": end.Sub(start).Milliseconds(),
|
||
"exception_info": exception,
|
||
"create_user": "daemon",
|
||
}).Insert(); werr != nil {
|
||
g.Log().Errorf(ctx, "任务日志写入失败 job=%d: %v", jobId, werr)
|
||
}
|
||
}
|
||
|
||
// truncStr 按字符数截断字符串
|
||
// 长度未超限时原样返回，超限时按 rune 截断，避免截断多字节字符
|
||
func truncStr(s string, maxLen int) string {
|
||
if len(s) <= maxLen {
|
||
return s
|
||
}
|
||
runes := []rune(s)
|
||
if len(runes) <= maxLen {
|
||
return s
|
||
}
|
||
return string(runes[:maxLen])
|
||
}
|