278 lines
10 KiB
Go
278 lines
10 KiB
Go
// +======================================================================
|
||
// | EasyGoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ]
|
||
// +----------------------------------------------------------------------
|
||
// | Copyright (c) 2019-2026 半城风雨团队 版权所有
|
||
// +----------------------------------------------------------------------
|
||
// | 开源协议: Apache-2.0
|
||
// | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。
|
||
// +----------------------------------------------------------------------
|
||
// | 官方网站: https://www.easygoadmin.vip
|
||
// +----------------------------------------------------------------------
|
||
// | 研发团队: 小蚂蚁云团队 荣誉出品
|
||
// +======================================================================
|
||
// | 版权和免责声明:
|
||
// | 一、知识产权声明
|
||
// | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、
|
||
// | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权，
|
||
// | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。
|
||
// | 二、使用限制与禁止条款
|
||
// | 您理解并同意，禁止将本产品用于以下任何行为：
|
||
// | 1. 任何违反中华人民共和国法律法规的用途；
|
||
// | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为；
|
||
// | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。
|
||
// | 三、责任豁免
|
||
// | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失
|
||
// | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三
|
||
// | 方提起索赔或行政机关进行调查，您应独立承担全部责任。
|
||
// | 四、违约与法律责任
|
||
// | 如发现任何违反本声明的行为，本团队有权：
|
||
// | 1. 无条件配合公安机关及其他有权机关的调查取证工作；
|
||
// | 2. 依法追究相关方的民事责任，并要求赔偿全部损失；
|
||
// | 3. 保留进一步采取刑事控告等法律措施的权利。
|
||
// | 五、许可范围说明
|
||
// | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包
|
||
// | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通
|
||
// | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、
|
||
// | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造
|
||
// | 成的一切经济损失及商誉损害。
|
||
// | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件
|
||
// | 许可协议》附件为准。
|
||
// +======================================================================
|
||
|
||
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])
|
||
}
|