perf: 优化定时任务

pull/148/head
xiaojunnuo 2024-08-05 16:00:04 +08:00
parent 2182dce07c
commit 87e440ee2a
6 changed files with 83 additions and 31 deletions

View File

@ -187,6 +187,9 @@ export default function ({ crudExpose, context: { certdFormRef } }: CreateCrudOp
title: "ID", title: "ID",
key: "id", key: "id",
type: "number", type: "number",
search: {
show: true
},
column: { column: {
width: 50 width: 50
}, },
@ -233,6 +236,31 @@ export default function ({ crudExpose, context: { certdFormRef } }: CreateCrudOp
} }
} }
}, },
content: {
title: "定时任务数量",
type: "number",
column: {
cellRender({ value }) {
if (value && value.triggers) {
return value.triggers?.length > 0 ? value.triggers.length : "-";
}
return "-";
}
},
valueBuilder({ row }) {
if (row.content) {
row.content = JSON.parse(row.content);
}
},
valueResolve({ row }) {
if (row.content) {
row.content = JSON.stringify(row.content);
}
},
form: {
show: false
}
},
lastVars: { lastVars: {
title: "到期剩余", title: "到期剩余",
type: "number", type: "number",

View File

@ -88,15 +88,16 @@ export class PipelineService extends BaseService<PipelineEntity> {
} }
async update(bean: PipelineEntity) { async update(bean: PipelineEntity) {
await this.clearTriggers(bean.id); //更新非trigger部分
await super.update(bean); await super.update(bean);
await this.registerTriggerById(bean.id);
} }
async save(bean: PipelineEntity) { async save(bean: PipelineEntity) {
await this.clearTriggers(bean.id); await this.clearTriggers(bean.id);
const pipeline = JSON.parse(bean.content); if (bean.content) {
bean.title = pipeline.title; const pipeline = JSON.parse(bean.content);
bean.title = pipeline.title;
}
await this.addOrUpdate(bean); await this.addOrUpdate(bean);
await this.registerTriggerById(bean.id); await this.registerTriggerById(bean.id);
} }

View File

@ -22,6 +22,7 @@ export class CronConfiguration {
...this.config, ...this.config,
}); });
container.registerObject('cron', this.cron); container.registerObject('cron', this.cron);
this.cron.start();
this.logger.info('cron started'); this.logger.info('cron started');
} }
} }

View File

@ -17,56 +17,64 @@ export class CronTask {
name: string; name: string;
stoped = false; stoped = false;
timeoutId: any; nextTime: any;
constructor(req: CronTaskReq, logger: ILogger) { constructor(req: CronTaskReq, logger: ILogger) {
this.cron = req.cron; this.cron = req.cron;
this.job = req.job; this.job = req.job;
this.name = req.name; this.name = req.name;
this.logger = logger; this.logger = logger;
this.start();
} }
start() { genNextTime() {
if (!this.cron) { if (!this.cron) {
return; return null;
} }
if (this.stoped) { if (this.stoped) {
return; return null;
} }
const interval = parser.parseExpression(this.cron); const interval = parser.parseExpression(this.cron);
const next = interval.next().getTime(); const next = interval.next().getTime();
const now = Date.now(); this.logger.info(`[cron] [${this.name}], cron:${this.cron}, next run :${new Date(next).toLocaleString()}`);
const delay = next - now; this.nextTime = next;
this.timeoutId = setTimeout(async () => { return next;
try {
if (this.stoped) {
return;
}
await this.job();
} catch (e) {
this.logger.error(`[cron] job error : [${this.name}]`, e);
}
this.start();
}, delay);
} }
stop() { stop() {
this.stoped = true; this.stoped = true;
clearTimeout(this.timeoutId);
} }
} }
export class Cron { export class Cron {
logger: ILogger; logger: ILogger;
immediateTriggerOnce: boolean; immediateTriggerOnce: boolean;
bucket: Record<string, CronTask> = {}; queue: CronTask[] = [];
constructor(opts: any) { constructor(opts: any) {
this.logger = opts.logger; this.logger = opts.logger;
this.immediateTriggerOnce = opts.immediateTriggerOnce; this.immediateTriggerOnce = opts.immediateTriggerOnce;
} }
start() {
this.logger.info('[cron] start');
this.queue.forEach(task => {
task.genNextTime();
});
setInterval(() => {
const now = new Date().getTime();
for (const task of this.queue) {
if (task.nextTime <= now) {
task.job().catch(e => {
this.logger.error(`job execute error : [${task.name}]`, e);
});
task.genNextTime();
} else {
break;
}
}
}, 1000 * 60);
}
register(req: CronTaskReq) { register(req: CronTaskReq) {
if (!req.cron) { if (!req.cron) {
this.logger.info(`[cron] register once : [${req.name}]`); this.logger.info(`[cron] register once : [${req.name}]`);
@ -78,21 +86,26 @@ export class Cron {
this.logger.info(`[cron] register cron : [${req.name}] ,${req.cron}`); this.logger.info(`[cron] register cron : [${req.name}] ,${req.cron}`);
const task = new CronTask(req, this.logger); const task = new CronTask(req, this.logger);
this.bucket[task.name] = task; task.genNextTime();
this.queue.push(task);
// sort by nextTime
this.queue.sort((a, b) => a.nextTime - b.nextTime);
this.logger.info('当前定时任务数量:', this.getTaskSize()); this.logger.info('当前定时任务数量:', this.getTaskSize());
} }
remove(taskName: string) { remove(taskName: string) {
this.logger.info(`[cron] remove : [${taskName}]`); this.logger.info(`[cron] remove : [${taskName}]`);
const task = this.bucket[taskName]; const index = this.queue.findIndex(item => item.name === taskName);
if (task) { if (index !== -1) {
task.stop(); this.queue[index].stop();
delete this.bucket[taskName]; this.queue.splice(index, 1);
} }
} }
getTaskSize() { getTaskSize() {
const tasks = Object.keys(this.bucket); const tasks = Object.keys(this.queue);
return tasks.length; return tasks.length;
} }
} }

View File

@ -75,6 +75,10 @@ export class AsyncSsh2Client {
} }
async exec(script: string) { async exec(script: string) {
if (!script) {
this.logger.info('script 为空,取消执行');
return;
}
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
this.logger.info(`执行命令:[${this.connConf.host}][exec]: ` + script); this.logger.info(`执行命令:[${this.connConf.host}][exec]: ` + script);
this.conn.exec(script, (err: Error, stream: any) => { this.conn.exec(script, (err: Error, stream: any) => {
@ -97,6 +101,10 @@ export class AsyncSsh2Client {
data += out; data += out;
this.logger.info(`[${this.connConf.host}][info]: ` + out.trimEnd()); this.logger.info(`[${this.connConf.host}][info]: ` + out.trimEnd());
}) })
.on('error', (err: any) => {
reject(err);
this.logger.error(err);
})
.stderr.on('data', (ret: Buffer) => { .stderr.on('data', (ret: Buffer) => {
const err = this.convert(ret); const err = this.convert(ret);
data += err; data += err;

View File

@ -30,6 +30,7 @@ export class HostShellExecutePlugin extends AbstractTaskPlugin {
name: 'a-textarea', name: 'a-textarea',
vModel: 'value', vModel: 'value',
}, },
required: true,
}) })
script!: string; script!: string;