chore: src/modules/task/service/local.ts
This commit is contained in:
@@ -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<TaskInfoEntity>;
|
||||
|
||||
@Logger()
|
||||
logger: ILogger;
|
||||
|
||||
@InjectEntityModel(TaskLogEntity)
|
||||
taskLogEntity: Repository<TaskLogEntity>;
|
||||
|
||||
@App()
|
||||
app: IMidwayApplication;
|
||||
|
||||
@Inject()
|
||||
utils: Utils;
|
||||
|
||||
@Config('task.log.keepDays')
|
||||
keepDays: number;
|
||||
|
||||
@Inject()
|
||||
coolEventManager: CoolEventManager;
|
||||
|
||||
// 存储所有运行的任务
|
||||
private cronJobs: Map<string, CronJob.CronJob> = 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<any> {
|
||||
const info = await this.taskInfoEntity.findOneBy({ id });
|
||||
return {
|
||||
...info,
|
||||
repeatCount: info.limit,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行器
|
||||
*/
|
||||
async executor(task: any): Promise<void> {
|
||||
// 如果不是开始时间之后的 则不执行
|
||||
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 });
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user