From 7fc298b4e8f2df1ec3837d93cbc79d83ba513a3e Mon Sep 17 00:00:00 2001 From: snowgitea Date: Wed, 1 Jul 2026 17:57:31 +0800 Subject: [PATCH] chore: src/modules/task/service/local.ts --- src/modules/task/service/local.ts | 351 ++++++++++++++++++++++++++++++ 1 file changed, 351 insertions(+) create mode 100644 src/modules/task/service/local.ts diff --git a/src/modules/task/service/local.ts b/src/modules/task/service/local.ts new file mode 100644 index 0000000..55e1d2a --- /dev/null +++ b/src/modules/task/service/local.ts @@ -0,0 +1,351 @@ +import { + App, + Config, + Inject, + Logger, + Provide, + Scope, + ScopeEnum, +} from '@midwayjs/core'; +import { BaseService, CoolEventManager } from '@cool-midway/core'; +import { InjectEntityModel } from '@midwayjs/typeorm'; +import { Equal, LessThan, Repository } from 'typeorm'; +import { TaskInfoEntity } from '../entity/info'; +import { TaskLogEntity } from '../entity/log'; +import { ILogger } from '@midwayjs/logger'; +import * as _ from 'lodash'; +import { Utils } from '../../../comm/utils'; +import { IMidwayApplication } from '@midwayjs/core'; +import { v4 as uuidv4 } from 'uuid'; +import * as moment from 'moment'; +import * as CronJob from 'cron'; + +/** + * 本地任务 + */ +@Provide() +@Scope(ScopeEnum.Singleton) +export class TaskLocalService extends BaseService { + @InjectEntityModel(TaskInfoEntity) + taskInfoEntity: Repository; + + @Logger() + logger: ILogger; + + @InjectEntityModel(TaskLogEntity) + taskLogEntity: Repository; + + @App() + app: IMidwayApplication; + + @Inject() + utils: Utils; + + @Config('task.log.keepDays') + keepDays: number; + + @Inject() + coolEventManager: CoolEventManager; + + // 存储所有运行的任务 + private cronJobs: Map = new Map(); + + /** + * 停止任务 + */ + async stop(id) { + const task = await this.taskInfoEntity.findOneBy({ id: Equal(id) }); + if (task) { + this.stopByJobId(task.jobId); + this.coolEventManager.emit('onLocalTaskStop', task.jobId); + task.status = 0; + await this.taskInfoEntity.update(task.id, task); + await this.updateNextRunTime(task.jobId); + } + } + + /** + * 停止任务 + * @param jobId + */ + async stopByJobId(jobId) { + const job = this.cronJobs.get(jobId); + if (job) { + job.stop(); + this.cronJobs.delete(jobId); + } + } + + /** + * 开始任务 + */ + async start(id, type?) { + const task = await this.taskInfoEntity.findOneBy({ id: Equal(id) }); + task.status = 1; + if (type || type == 0) { + task.type = type; + } + await this.addOrUpdate(task); + } + + /** + * 手动执行一次 + */ + async once(id) { + const task = await this.taskInfoEntity.findOneBy({ id: Equal(id) }); + if (task) { + await this.executeJob(task); + } + } + + /** + * 检查任务是否存在 + */ + async exist(jobId) { + return this.cronJobs.has(jobId); + } + + /** + * 创建定时任务 + */ + private createCronJob(task) { + let cronTime; + if (task.taskType === 0) { + // cron 类型 + cronTime = task.cron; + } else { + // 间隔类型 + cronTime = `*/${task.every / 1000} * * * * *`; + } + + const job = new CronJob.CronJob( + cronTime, + async () => { + await this.executeJob(task); + }, + null, + false, + 'Asia/Shanghai' + ); + + this.cronJobs.set(task.jobId, job); + job.start(); + return job; + } + + /** + * 执行任务 + */ + private async executeJob(task) { + await this.executor(task); + } + + /** + * 新增或修改 + */ + async addOrUpdate(params) { + if (!params.jobId) { + params.jobId = uuidv4(); + } + + await this.getOrmManager().transaction(async transactionalEntityManager => { + if (params.taskType === 0) { + params.limit = null; + params.every = null; + } else { + params.cron = null; + } + await transactionalEntityManager.save(TaskInfoEntity, params); + + if (params.status === 1) { + const exist = await this.exist(params.jobId); + if (exist) { + const job = this.cronJobs.get(params.jobId); + job.stop(); + this.cronJobs.delete(params.jobId); + this.coolEventManager.emit('onLocalTaskStop', params.jobId); + } + this.createCronJob(params); + } + }); + + if (params.status === 1) { + await this.updateNextRunTime(params.jobId); + } + } + + /** + * 删除任务 + */ + async delete(ids) { + let idArr; + if (ids instanceof Array) { + idArr = ids; + } else { + idArr = ids.split(','); + } + for (const id of idArr) { + const task = await this.taskInfoEntity.findOneBy({ id }); + if (task) { + const job = this.cronJobs.get(task.jobId); + if (job) { + job.stop(); + this.cronJobs.delete(task.jobId); + } + await this.taskInfoEntity.delete({ id }); + await this.taskLogEntity.delete({ taskId: id }); + } + } + } + + /** + * 记录任务执行情况 + */ + async record(task, status, detail?) { + const info = await this.taskInfoEntity.findOneBy({ + jobId: Equal(task.jobId), + }); + await this.taskLogEntity.save({ + taskId: info.id, + status, + detail: detail || '', + }); + await this.taskLogEntity.delete({ + taskId: info.id, + createTime: LessThan(moment().subtract(this.keepDays, 'days').toDate()), + }); + } + + /** + * 获取下次执行时间 + */ + async getNextRunTime(jobId) { + const job = this.cronJobs.get(jobId); + return job ? job.nextDate().toJSDate() : null; + } + + /** + * 更新下次执行时间 + */ + async updateNextRunTime(jobId) { + const nextRunTime = await this.getNextRunTime(jobId); + if (nextRunTime) { + await this.taskInfoEntity.update({ jobId }, { nextRunTime }); + } + } + + /** + * 初始化任务 + */ + async initTask() { + try { + this.logger.info('init local task....'); + const runningTasks = await this.taskInfoEntity.findBy({ status: 1 }); + if (!_.isEmpty(runningTasks)) { + for (const task of runningTasks) { + const job = await this.exist(task.jobId); + if (!job) { + this.logger.info(`init local task ${task.name}`); + await this.addOrUpdate(task); + } + } + } + } catch (e) { + this.logger.error('Init local task error:', e); + } + } + + /** + * 调用service + */ + async invokeService(serviceStr) { + if (serviceStr) { + const arr = serviceStr.split('.'); + const service = await this.app + .getApplicationContext() + .getAsync(_.lowerFirst(arr[0])); + + for (let i = 1; i < arr.length; i++) { + const child = arr[i]; + if (child.includes('(')) { + const [methodName, paramsStr] = child.split('('); + const params = paramsStr + .replace(')', '') + .split(',') + .map(param => param.trim()); + + if (params.length === 1 && params[0] === '') { + return service[methodName](); + } else { + const parsedParams = params.map(param => { + try { + return JSON.parse(param); + } catch (e) { + return param; + } + }); + return service[methodName](...parsedParams); + } + } + } + } + } + + /** + * 获取任务详情 + */ + async info(id: any): Promise { + const info = await this.taskInfoEntity.findOneBy({ id }); + return { + ...info, + repeatCount: info.limit, + }; + } + + /** + * 执行器 + */ + async executor(task: any): Promise { + // 如果不是开始时间之后的 则不执行 + if (task.startDate && moment(task.startDate).isAfter(moment())) { + return; + } + try { + const currentTime = moment().toDate(); + const lockExpireTime = moment().add(5, 'minutes').toDate(); + const result = await this.taskInfoEntity + .createQueryBuilder() + .update() + .set({ + lastExecuteTime: currentTime, + lockExpireTime: lockExpireTime, + }) + .where('id = :id', { id: task.id }) + .andWhere('lockExpireTime IS NULL OR lockExpireTime < :currentTime', { + currentTime, + }) + .execute(); + + // 如果更新失败（affected === 0），说明其他实例正在执行 + if (result.affected === 0) { + return; + } + + const serviceResult = await this.invokeService(task.service); + await this.record(task, 1, JSON.stringify(serviceResult)); + } catch (error) { + await this.record(task, 0, error.message); + } finally { + // 释放锁 + await this.taskInfoEntity.update( + { id: task.id }, + { lockExpireTime: null } + ); + } + + if (!task.isOnce) { + await this.updateNextRunTime(task.jobId); + await this.taskInfoEntity.update({ id: task.id }, { status: 1 }); + } + } +}