import { Provide, Inject } from '@midwayjs/core';
|
import { InjectEntityModel } from '@midwayjs/typeorm';
|
import { Repository, In } from 'typeorm';
|
import { ReportJoblogEntity } from '../entity/joblog';
|
import { TaskServiceSorter } from './taskservicesorter'
|
|
|
|
@Provide()
|
export class ExecutorSorterDaily {
|
|
@InjectEntityModel(ReportJoblogEntity)
|
reportJoblogEntity: Repository<ReportJoblogEntity>;
|
|
@Inject()
|
taskServiceSorter: TaskServiceSorter;
|
|
async execute() {
|
// 捞出所有待处理任务
|
const pendingSiteJobs = await this.reportJoblogEntity.find({
|
where: {
|
status: In(['INIT', 'FAILED']),
|
jobName: In(['SORTER_DAILY']),
|
},
|
order: {
|
dataDate: 'ASC',
|
},
|
});
|
|
for (const job of pendingSiteJobs) {
|
|
// 锁定任务状态为 RUNNING（轻量级乐观锁锁机制）
|
const lockResult = await this.reportJoblogEntity.update(
|
{ id: job.id, status: job.status }, // 带有原始状态检查，防止被别的线程抢跑
|
{ status: 'RUNNING', startTime: new Date() }
|
);
|
|
if (lockResult.affected === 0) {
|
continue; // 没抢到锁，说明别的定时器在跑它，跳过
|
}
|
|
// 真正开始跑数据
|
try {
|
let recordsProcessed = 0;
|
|
|
recordsProcessed = await this.taskServiceSorter.aggregateDailyData(job.dataDate) || 0;
|
|
|
// 成功：更新状态
|
await this.reportJoblogEntity.update(job.id, {
|
status: 'SUCCESS',
|
recordsProcessed: recordsProcessed,
|
endTime: new Date(),
|
errorMessage: '',
|
});
|
|
} catch (error) {
|
// 失败：记录堆栈，等待下一小时重试
|
|
var errorMessage = '';
|
if (error instanceof Error) {
|
errorMessage = error.stack ?? error.message;
|
}
|
if (typeof error === 'string') {
|
errorMessage = error;
|
}
|
try {
|
errorMessage = JSON.stringify(error);
|
} catch {
|
errorMessage = String(error);
|
}
|
|
await this.reportJoblogEntity.update(job.id, {
|
status: 'FAILED',
|
endTime: new Date(),
|
errorMessage: errorMessage,
|
failedCount: () => 'failedCount + 1',
|
});
|
}
|
}
|
}
|
|
}
|
