import { Provide, App, Scope, ScopeEnum } from "@midwayjs/core"; import { Application } from "@midwayjs/koa"; import { InjectRepository } from "@midwayjs/sequelize"; import { CronJob, CronTime } from "cron"; import { Op } from "sequelize"; import dayjs from "dayjs"; import SysJob from "../entity/sys_job.entity"; import SysJobLog from "../entity/sys_job_log.entity"; import { LogService } from "./log.service"; import { NoticeRuleService } from "./notice_rule.service"; import { IccLineSyncService } from "./icc_line_sync.service"; import { IccCustomerSyncService } from "./icc_customer_sync.service"; import { BuiltinJob, JobHandlerContext, JobRuntime, getBuiltinJob } from "../task/job_registry"; /** 任务执行结果(一次尝试) */ export interface JobRunResult { status: number; // 1成功 2失败 3跳过 message: string; // 结果摘要 costMs: number; // 本次尝试耗时 attempt: number; // 本次是第几次尝试 } /** 调度时区(固定上海时区) */ const TIMEZONE = "Asia/Shanghai"; /** 异常堆栈最大存储长度 */ const MAX_ERROR_STACK = 4000; /** * 自研轻量调度器:任务清单来自代码注册表,调度配置持久化在 sys_job, * 页面修改 cron / 启停后通过 reload 即时生效(无需重启)。 */ // 调度器必须是进程内单例:定时器注册表要能被「启动注册」和「接口重载」共享 @Provide() @Scope(ScopeEnum.Singleton) export class JobSchedulerService { @App() app!: Application; @InjectRepository(SysJob) jobModel!: typeof SysJob; @InjectRepository(SysJobLog) jobLogModel!: typeof SysJobLog; /** jobKey → 定时器 */ private timers = new Map(); /** jobKey → 待执行的重试定时器 */ private retryTimers = new Map(); /** jobKey 重入锁 */ private running = new Set(); /** 私有日志适配:注册表与任务经它输出日志,沿用仓库内 console 风格 */ private readonly jobLogger = { info: (m: string, ...a: any[]) => console.log(`[定时任务] ${m}`, ...a), warn: (m: string, ...a: any[]) => console.warn(`[定时任务] ${m}`, ...a), error: (m: string, ...a: any[]) => console.error(`[定时任务] ${m}`, ...a) }; /** 应用启动时加载全部启用任务并注册调度;失败自动重试,最多 3 次,不阻塞启动 */ async start(): Promise { const maxAttempts = 3; for (let attempt = 1; attempt <= maxAttempts; attempt++) { try { const jobs = await this.jobModel.findAll({ where: { status: 1 } }); jobs.forEach(job => this.register(job)); this.jobLogger.info(`调度器启动完成,已加载 ${jobs.length} 个任务`); return; } catch (err) { if (attempt < maxAttempts) { this.jobLogger.error(`调度器启动失败(第 ${attempt} 次尝试),5 秒后重试:`, err); await new Promise(resolve => setTimeout(resolve, 5000)); } else { // 3 次仍失败仅记录,不抛出,避免阻塞应用启动 console.error("[定时任务] 调度器启动失败,已达最大重试次数,应用继续启动:", err); } } } } /** 停止全部定时器(应用停止时调用) */ async stop(): Promise { for (const timer of this.timers.values()) { try { timer.stop(); } catch (err) { console.warn("[定时任务] 停止定时器失败:", err); } } this.timers.clear(); for (const retryTimer of this.retryTimers.values()) { clearTimeout(retryTimer); } this.retryTimers.clear(); } /** 单个任务重新加载:停旧 → 按最新配置重新注册(停用则不注册) */ async reload(jobKey: string): Promise { this.unregister(jobKey); const job = await this.jobModel.findOne({ where: { jobKey } }); if (!job || Number(job.status) !== 1) { return; } this.register(job); } /** 立即执行一次入口(手动触发同样按配置重试,但不阻塞调用方) */ async runOnce(jobId: number, triggerType: number): Promise { return this.execute(jobId, triggerType); } /** 校验 cron 表达式合法性 */ validateCron(expression: string): boolean { if (!expression || typeof expression !== "string") { return false; } try { new CronTime(expression, TIMEZONE); return true; } catch { return false; } } /** 计算下次执行时间;表达式非法返回 null */ nextRunTime(cronExpression: string): Date | null { if (!this.validateCron(cronExpression)) { return null; } try { // luxon 3.x 中 toDate 已移除,使用 toJSDate return new CronTime(cronExpression, TIMEZONE).sendAt().toJSDate(); } catch { return null; } } /** 计算未来 count 次执行时间(配置界面预览用);表达式非法返回空数组 */ nextRunTimes(cronExpression: string, count: number): Date[] { if (!this.validateCron(cronExpression)) { return []; } try { return new CronTime(cronExpression, TIMEZONE) .sendAt(count) .map(date => date.toJSDate()); } catch { return []; } } /** 清理 sys_job_log 中早于 keepDays 天的记录,返回删除条数 */ async cleanJobLogs(keepDays: number): Promise { const cutoff = dayjs().subtract(keepDays, "day").toDate(); return this.jobLogModel.destroy({ where: { created_at: { [Op.lt]: cutoff } } }); } /** 按任务配置注册定时器 */ private register(job: SysJob): void { this.unregister(job.jobKey); const builtin = getBuiltinJob(job.jobKey); if (!builtin) { this.jobLogger.warn(`任务 ${job.jobKey} 未在注册表中实现,跳过调度注册`); return; } if (!this.validateCron(job.cronExpression)) { this.jobLogger.warn(`任务 ${job.jobKey} 的 cron 表达式不合法(${job.cronExpression}),跳过调度注册`); return; } try { const timer = new CronJob( job.cronExpression, () => { void this.execute(job.id, 1); }, null, true, TIMEZONE ); this.timers.set(job.jobKey, timer); this.jobLogger.info(`任务 ${job.jobKey} 已注册调度,cron=${job.cronExpression}`); } catch (err) { this.jobLogger.warn(`任务 ${job.jobKey} 定时器创建失败:`, err); } } /** 注销定时器与待执行的重试定时器(不清重入锁) */ private unregister(jobKey: string): void { const timer = this.timers.get(jobKey); if (timer) { try { timer.stop(); } catch (err) { console.warn(`[定时任务] 停止任务 ${jobKey} 定时器失败:`, err); } this.timers.delete(jobKey); } const retryTimer = this.retryTimers.get(jobKey); if (retryTimer) { clearTimeout(retryTimer); this.retryTimers.delete(jobKey); } } /** 执行内核:重入保护 → 执行尝试 → 失败按配置进入重试链 */ private async execute(jobId: number, triggerType: number): Promise { const job = await this.jobModel.findByPk(jobId); if (!job) { return { status: 2, message: "任务不存在", costMs: 0, attempt: 0 }; } const builtin = getBuiltinJob(job.jobKey); if (!builtin) { const now = new Date(); await this.writeJobLog({ job, triggerType, status: 2, attempt: 1, startedAt: now, finishedAt: now, costMs: 0, message: "任务实现不存在", errorStack: null }); await this.writeLastRun(job, { status: 2, message: "任务实现不存在", costMs: 0, attempt: 1 }); return { status: 2, message: "任务实现不存在", costMs: 0, attempt: 1 }; } // 并发保护:上一次尚未结束则本次跳过,且不改写 sys_job.last_* if (this.running.has(job.jobKey)) { const message = "上一次执行尚未结束,本次跳过"; const now = new Date(); await this.writeJobLog({ job, triggerType, status: 3, attempt: 1, startedAt: now, finishedAt: now, costMs: 0, message, errorStack: null }); return { status: 3, message, costMs: 0, attempt: 1 }; } this.running.add(job.jobKey); // 失败且需重试时把锁移交给重试链,本次不释放 let handedOverToRetry = false; try { const result = await this.runAttempt(job, builtin, triggerType, 1); if (result.status !== 1 && Number(job.retryTimes) > 0) { handedOverToRetry = true; this.scheduleRetry(job, 2); } else { await this.writeLastRun(job, result); } return { status: result.status, message: result.message, costMs: result.costMs, attempt: result.attempt }; } catch (err) { const e = err as any; this.jobLogger.error(`任务 ${job.jobKey} 执行流程异常:`, err); return { status: 2, message: e?.message ? String(e.message) : "执行失败", costMs: 0, attempt: 1 }; } finally { if (!handedOverToRetry) { this.running.delete(job.jobKey); } } } /** 重试链:失败重试(trigger_type = 3),重试期间保持重入锁 */ private async continueChain(jobId: number, attempt: number, jobKeyHint?: string): Promise { let jobKey = jobKeyHint; let handedOverToRetry = false; try { const job = await this.jobModel.findByPk(jobId); if (!job) { return; } jobKey = job.jobKey; const builtin = getBuiltinJob(job.jobKey); if (!builtin) { const now = new Date(); await this.writeJobLog({ job, triggerType: 3, status: 2, attempt, startedAt: now, finishedAt: now, costMs: 0, message: "任务实现不存在", errorStack: null }); await this.writeLastRun(job, { status: 2, message: "任务实现不存在", costMs: 0, attempt }); return; } const result = await this.runAttempt(job, builtin, 3, attempt); if (result.status !== 1 && attempt <= Number(job.retryTimes)) { handedOverToRetry = true; this.scheduleRetry(job, attempt + 1); return; } // 重试链结束(成功或超出重试次数)才回写任务的 last_run_* await this.writeLastRun(job, result); } catch (err) { this.jobLogger.error(`任务重试执行异常:`, err); } finally { if (!handedOverToRetry && jobKey) { this.running.delete(jobKey); } } } /** 单次尝试:执行 handler 并写入一条执行日志 */ private async runAttempt( job: SysJob, builtin: BuiltinJob, triggerType: number, attempt: number ): Promise { const startedAt = new Date(); const startMs = Date.now(); let status = 1; let message = "执行成功"; let errorStack: string | null = null; let error: any; try { const ctx: JobHandlerContext = { jobKey: job.jobKey, jobName: job.jobName, triggerType, attempt, runtime: this.buildRuntime() }; const ret = await builtin.handler(job.params ?? {}, ctx); message = typeof ret === "string" && ret ? ret : "执行成功"; } catch (err) { const e = err as any; status = 2; message = e?.message ? String(e.message) : "执行失败"; errorStack = (e?.stack ? String(e.stack) : String(e)).slice(0, MAX_ERROR_STACK); error = err; } const finishedAt = new Date(); const costMs = Date.now() - startMs; await this.writeJobLog({ job, triggerType, status, attempt, startedAt, finishedAt, costMs, message, errorStack }); return { status, message, costMs, attempt, error }; } /** 写入一条执行日志(写入失败不影响任务执行) */ private async writeJobLog(fields: { job: SysJob; triggerType: number; status: number; attempt: number; startedAt: Date; finishedAt: Date; costMs: number; message: string; errorStack: string | null; }): Promise { try { await this.jobLogModel.create({ jobId: fields.job.id, jobKey: fields.job.jobKey, jobName: fields.job.jobName, triggerType: fields.triggerType, status: fields.status, attempt: fields.attempt, startedAt: fields.startedAt, finishedAt: fields.finishedAt, costMs: fields.costMs, message: fields.message, errorStack: fields.errorStack } as any); } catch (err) { this.jobLogger.error(`任务 ${fields.job.jobKey} 执行日志写入失败:`, err); } } /** 回写任务上次执行信息 */ private async writeLastRun(job: SysJob, result: JobRunResult): Promise { await this.jobModel.update( { lastRunAt: new Date(), lastRunStatus: result.status, lastRunCostMs: result.costMs, lastError: result.status === 1 ? null : result.message } as any, { where: { id: job.id } } ); } /** 排定一次重试(重试期间保持重入锁) */ private scheduleRetry(job: SysJob, attempt: number): void { const intervalSec = Math.max(10, Number(job.retryInterval) || 60); const retryTimer = setTimeout(() => { this.retryTimers.delete(job.jobKey); void this.continueChain(job.id, attempt, job.jobKey); }, intervalSec * 1000); this.retryTimers.set(job.jobKey, retryTimer); } /** 构造注入给任务的运行时能力 */ private buildRuntime(): JobRuntime { return { logger: this.jobLogger, // LogService 是请求作用域,单例中改为按需从匿名请求上下文获取 cleanOperationLogs: async (keepDays: number) => { const ctx = this.app.createAnonymousContext(); const logService = (await ctx.requestContext.getAsync(LogService)) as LogService; return logService.cleanExpiredLogs(keepDays); }, cleanJobLogs: (keepDays: number) => this.cleanJobLogs(keepDays), // NoticeRuleService 同样是请求作用域,按需从匿名请求上下文获取 runNoticeRules: async () => { const ctx = this.app.createAnonymousContext(); const noticeRuleService = (await ctx.requestContext.getAsync(NoticeRuleService)) as NoticeRuleService; return noticeRuleService.executeScan(); }, syncIccLines: async () => { const ctx = this.app.createAnonymousContext(); const iccService = (await ctx.requestContext.getAsync(IccLineSyncService)) as IccLineSyncService; const result = await iccService.syncFromIcc(); return `拉取 ${result.pulled} 条,新增 ${result.inserted},更新 ${result.updated},跳过 ${result.skipped}`; }, syncIccCustomers: async () => { const ctx = this.app.createAnonymousContext(); const iccCustomerService = (await ctx.requestContext.getAsync(IccCustomerSyncService)) as IccCustomerSyncService; const result = await iccCustomerService.syncFromIcc(); return `拉取 ${result.pulled} 条,新增 ${result.inserted},更新 ${result.updated},跳过 ${result.skipped}`; } }; } }