tg_admin_server/app/service/admin/sys/task.ts

import BaseService from '../../base';
import { CreateTaskDto, UpdateTaskDto } from '../../../dto/admin/sys/task';
import SysTask from '../../../entities/admin/sys/task';

/**
 * 任务调度服务
 */
export default class SysTaskService extends BaseService {

  /**
   * 初始化任务，系统启动前调用
   */
  async initTask() {
    const jobs = await this.app.queue.sys.getJobs([ 'active', 'delayed', 'failed', 'paused', 'waiting', 'completed' ]);
    for (let i = 0; i < jobs.length; i++) {
      // 先移除所有已存在的任务
      await jobs[i].remove();
    }
    // 查找所有需要运行的任务
    const tasks = await this.getRepo().admin.sys.Task.find({ status: 1 });
    if (tasks && tasks.length > 0) {
      for (const t of tasks) {
        this.start(t);
      }
    }
  }

  /**
   * 分页查询
   */
  async page(page: number, count: number) {
    const result = await this.getRepo().admin.sys.Task.find({
      order: {
        id: 'ASC',
      },
      take: count,
      skip: page * count,
    });
    return result;
  }

  /**
   * count task
   */
  async count() {
    return await this.getRepo().admin.sys.Task.count();
  }

  /**
   * task info
   */
  async info(id: number) {
    return await this.getRepo().admin.sys.Task.findOne({ id });
  }

  /**
   * delete task
   */
  async delete(task: SysTask) {
    if (!task) {
      throw new Error('Task is Empty');
    }
    await this.stop(task);
    await this.getRepo().admin.sys.Task.delete(task.id);
  }

  /**
   * 添加任务
   */
  async addOrUpdate(param: CreateTaskDto | UpdateTaskDto) {
    const result = await this.getRepo().admin.sys.Task.save(param);
    const task = await this.info(result.id);
    if (result.status === 0) {
      await this.stop(task!);
    } else if (result.status === 1) {
      await this.start(task!);
    }
  }

  /**
   * 手动执行一次
   */
  async once(task: SysTask) {
    if (task) {
      await this.app.queue.sys.add({ id: task.id, service: task.service, args: task.data },
        { jobId: task.id, removeOnComplete: true, removeOnFail: true });
    } else {
      throw new Error('Task is Empty');
    }
  }

  /**
   * 启动任务
   */
  async start(task: SysTask) {
    if (!task) {
      throw new Error('Task is Empty');
    }
    // 先停掉之前存在的任务
    await this.stop(task);
    let repeat: any;
    if (task.type === 1) {
      // 间隔 Repeat every millis (cron setting cannot be used together with this setting.)
      repeat = {
        every: task.every,
      };
    } else {
      // cron
      repeat = {
        cron: task.cron,
      };
      // Start date when the repeat job should start repeating (only with cron).
      if (task.startTime) {
        repeat.startDate = task.startTime;
      }
      if (task.endTime) {
        repeat.endDate = task.endTime;
      }
    }
    if (task.limit > 0) {
      repeat.limit = task.limit;
    }
    const job = await this.app.queue.sys.add({ id: task.id, service: task.service, args: task.data },
      { jobId: task.id, removeOnComplete: true, removeOnFail: true, repeat });
    if (job && job.opts) {
      await this.getRepo().admin.sys.Task.update(task.id, { jobOpts: JSON.stringify(job.opts.repeat), status: 1 });
    } else {
      // update status to 0，标识暂停任务，因为启动失败
      job && await job.remove();
      await this.getRepo().admin.sys.Task.update(task.id, { status: 0 });
      throw new Error('Task Start failed');
    }
  }

  /**
   * 停止任务
   */
  async stop(task: SysTask) {
    if (!task) {
      throw new Error('Task is Empty');
    }
    const exist = await this.existJob(task.id.toString());
    if (!exist) {
      await this.getRepo().admin.sys.Task.update(task.id, { status: 0 });
      return;
    }
    const jobs = await this.app.queue.sys.getJobs([ 'active', 'delayed', 'failed', 'paused', 'waiting', 'completed' ]);
    for (let i = 0; i < jobs.length; i++) {
      if (jobs[i].data.id === task.id) {
        await jobs[i].remove();
      }
    }
    await this.getRepo().admin.sys.Task.update(task.id, { status: 0 });
    // if (task.jobOpts) {
    //   await this.app.queue.sys.removeRepeatable(JSON.parse(task.jobOpts));
    //   // update status
    //   await this.getRepo().admin.sys.Task.update(task.id, { status: 0 });
    // }
  }

  /**
   * 查看队列中任务是否存在
   */
  async existJob(jobId: string) {
    const jobs = await this.app.queue.sys.getRepeatableJobs();
    const ids = jobs.map(e => {
      return e.id;
    });
    return ids.includes(jobId);
  }

  /**
   * 更新任务完成状态
   */
  async updateTaskCompleteStatus(tid: number) {
    const result = await this.app.queue.sys.getRepeatableJobs();
    const task = await this.getRepo().admin.sys.Task.findOne({ id: tid });
    const jobOpts = JSON.parse(task!.jobOpts);
    // 如果下次执行时间小于当前时间，则表示已经执行完成。
    for (const task of result) {
      const currentTime = new Date().getTime();
      if (task.id && task.id === tid.toString()
        && (jobOpts.cron ? task.cron === jobOpts.cron : task.every === jobOpts.every) && task.next < currentTime) {
        // 如果下次执行时间小于当前时间，则表示已经执行完成。
        await this.getRepo().admin.sys.Task.update(tid, { status: 2 });
      }
    }
  }

  /**
   * 根据serviceName调用service
   */
  async callService(serviceName: string, args: string) {
    if (serviceName) {
      let serviceTmp = this.service;
      const arr = serviceName.split('.');
      for (let i = 0; i < arr.length; i++) {
        if (i === arr.length - 1) {
          if (args) {
            await serviceTmp[arr[arr.length - 1]](JSON.parse(args));
          } else {
            await serviceTmp[arr[arr.length - 1]]();
          }
        } else {
          serviceTmp = serviceTmp[arr[i]];
        }
      }
    }
  }

}