重构任务并发执行逻辑

back/schedule/addCron.ts

@@ -2,7 +2,7 @@ import { ServerUnaryCall, sendUnaryData } from '@grpc/grpc-js';
 import { AddCronRequest, AddCronResponse } from '../protos/cron';
 import nodeSchedule from 'node-schedule';
 import { scheduleStacks } from './data';
-import { exec } from 'child_process';
+import { runCron } from '../shared/runCron';
 
 const addCron = (
   call: ServerUnaryCall<AddCronRequest, AddCronResponse>,
@@ -16,7 +16,7 @@ const addCron = (
     scheduleStacks.set(
       id,
       nodeSchedule.scheduleJob(id, schedule, async () => {
-        exec(`ID=${id} ${command}`);
+        runCron(`ID=${id} ${command}`)
       }),
     );
   }

back/schedule/health.ts

@@ -1,6 +1,5 @@
 import { ServerUnaryCall, sendUnaryData } from '@grpc/grpc-js';
 import { HealthCheckRequest, HealthCheckResponse } from '../protos/health';
-import { exec } from 'child_process';
 import config from '../config';
 import { promiseExec } from '../config/util';
 

back/services/cron.ts

@@ -16,6 +16,7 @@ import { Op, where, col as colFn, FindOptions } from 'sequelize';
 import path from 'path';
 import { TASK_PREFIX, QL_PREFIX } from '../config/const';
 import cronClient from '../schedule/client';
+import { runCronWithLimit } from '../shared/pLimit';
 
 @Service()
 export default class CronService {
@@ -365,10 +366,9 @@ export default class CronService {
       { status: CrontabStatus.queued },
       { where: { id: ids } },
     );
-    concurrentRun(
+    ids.forEach(id => {
-      ids.map((id) => async () => await this.runSingle(id)),
+      this.runSingle(id)
-      10,
+    })
-    );
   }
 
   public async stop(ids: number[]) {
@@ -390,65 +390,67 @@ export default class CronService {
   }
 
   private async runSingle(cronId: number): Promise<number> {
-    return new Promise(async (resolve: any) => {
+    return runCronWithLimit(() => {
-      const cron = await this.getDb({ id: cronId });
+      return new Promise(async (resolve: any) => {
-      if (cron.status !== CrontabStatus.queued) {
+        const cron = await this.getDb({ id: cronId });
-        resolve();
+        if (cron.status !== CrontabStatus.queued) {
-        return;
+          resolve();
-      }
+          return;
-
-      let { id, command, log_path } = cron;
-      const absolutePath = path.resolve(config.logPath, `${log_path}`);
-      const logFileExist = log_path && (await fileExist(absolutePath));
-
-      this.logger.silly('Running job');
-      this.logger.silly('ID: ' + id);
-      this.logger.silly('Original command: ' + command);
-
-      let cmdStr = command;
-      if (!cmdStr.startsWith(TASK_PREFIX) && !cmdStr.startsWith(QL_PREFIX)) {
-        cmdStr = `${TASK_PREFIX}${cmdStr}`;
-      }
-      if (
-        cmdStr.endsWith('.js') ||
-        cmdStr.endsWith('.py') ||
-        cmdStr.endsWith('.pyc') ||
-        cmdStr.endsWith('.sh') ||
-        cmdStr.endsWith('.ts')
-      ) {
-        cmdStr = `${cmdStr} now`;
-      }
-
-      const cp = spawn(`ID=${id} ${cmdStr}`, { shell: '/bin/bash' });
-
-      await CrontabModel.update(
-        { status: CrontabStatus.running, pid: cp.pid },
-        { where: { id } },
-      );
-      cp.stderr.on('data', (data) => {
-        if (logFileExist) {
-          fs.appendFileSync(`${absolutePath}`, `${data.toString()}`);
         }
-      });
-      cp.on('error', (err) => {
-        if (logFileExist) {
-          fs.appendFileSync(`${absolutePath}`, `${JSON.stringify(err)}`);
-        }
-      });
 
-      cp.on('exit', async (code, signal) => {
+        let { id, command, log_path } = cron;
-        this.logger.info(
+        const absolutePath = path.resolve(config.logPath, `${log_path}`);
-          `任务 ${command} 进程id: ${cp.pid} 退出，退出码 ${code}`,
+        const logFileExist = log_path && (await fileExist(absolutePath));
-        );
+
-      });
+        this.logger.silly('Running job');
-      cp.on('close', async (code) => {
+        this.logger.silly('ID: ' + id);
+        this.logger.silly('Original command: ' + command);
+
+        let cmdStr = command;
+        if (!cmdStr.startsWith(TASK_PREFIX) && !cmdStr.startsWith(QL_PREFIX)) {
+          cmdStr = `${TASK_PREFIX}${cmdStr}`;
+        }
+        if (
+          cmdStr.endsWith('.js') ||
+          cmdStr.endsWith('.py') ||
+          cmdStr.endsWith('.pyc') ||
+          cmdStr.endsWith('.sh') ||
+          cmdStr.endsWith('.ts')
+        ) {
+          cmdStr = `${cmdStr} now`;
+        }
+
+        const cp = spawn(`ID=${id} ${cmdStr}`, { shell: '/bin/bash' });
+
         await CrontabModel.update(
-          { status: CrontabStatus.idle, pid: undefined },
+          { status: CrontabStatus.running, pid: cp.pid },
           { where: { id } },
         );
-        resolve();
+        cp.stderr.on('data', (data) => {
+          if (logFileExist) {
+            fs.appendFileSync(`${absolutePath}`, `${data.toString()}`);
+          }
+        });
+        cp.on('error', (err) => {
+          if (logFileExist) {
+            fs.appendFileSync(`${absolutePath}`, `${JSON.stringify(err)}`);
+          }
+        });
+
+        cp.on('exit', async (code, signal) => {
+          this.logger.info(
+            `任务 ${command} 进程id: ${cp.pid} 退出，退出码 ${code}`,
+          );
+        });
+        cp.on('close', async (code) => {
+          await CrontabModel.update(
+            { status: CrontabStatus.idle, pid: undefined },
+            { where: { id } },
+          );
+          resolve();
+        });
       });
-    });
+    })
   }
 
   public async disabled(ids: number[]) {

back/services/dependence.ts

@@ -14,13 +14,14 @@ import SockService from './sock';
 import { FindOptions, Op } from 'sequelize';
 import { concurrentRun } from '../config/util';
 import dayjs from 'dayjs';
+import { runCronWithLimit } from 'back/shared/pLimit';
 
 @Service()
 export default class DependenceService {
   constructor(
     @Inject('logger') private logger: winston.Logger,
     private sockService: SockService,
-  ) {}
+  ) { }
 
   public async create(payloads: Dependence[]): Promise<Dependence[]> {
     const tabs = payloads.map((x) => {
@@ -104,20 +105,17 @@ export default class DependenceService {
     isInstall: boolean = true,
     force: boolean = false,
   ) {
-    concurrentRun(
+    docs.forEach(async (dep) => {
-      docs.map((dep) => async () => {
+      const status = isInstall
-        const status = isInstall
+        ? DependenceStatus.installing
-          ? DependenceStatus.installing
+        : DependenceStatus.removing;
-          : DependenceStatus.removing;
+      await DependenceModel.update({ status }, { where: { id: dep.id } });
-        await DependenceModel.update({ status }, { where: { id: dep.id } });
+      this.installOrUninstallDependencies(
-        return await this.installOrUninstallDependencies(
+        [dep],
-          [dep],
+        isInstall,
-          isInstall,
+        force,
-          force,
+      )
-        );
+    })
-      }),
-      1,
-    );
   }
 
   public async reInstall(ids: number[]): Promise<Dependence[]> {
@@ -157,72 +155,28 @@ export default class DependenceService {
     isInstall: boolean = true,
     force: boolean = false,
   ) {
-    return new Promise(async (resolve) => {
+    return runCronWithLimit(() => {
-      if (dependencies.length === 0) {
+      return new Promise(async (resolve) => {
-        resolve(null);
+        if (dependencies.length === 0) {
-        return;
+          resolve(null);
-      }
+          return;
-      const socketMessageType = !force
+        }
-        ? 'installDependence'
+        const socketMessageType = !force
-        : 'uninstallDependence';
+          ? 'installDependence'
-      const depNames = dependencies.map((x) => x.name).join(' ');
+          : 'uninstallDependence';
-      const depRunCommand = (
+        const depNames = dependencies.map((x) => x.name).join(' ');
-        isInstall
+        const depRunCommand = (
-          ? InstallDependenceCommandTypes
+          isInstall
-          : unInstallDependenceCommandTypes
+            ? InstallDependenceCommandTypes
-      )[dependencies[0].type as any];
+            : unInstallDependenceCommandTypes
-      const actionText = isInstall ? '安装' : '删除';
+        )[dependencies[0].type as any];
-      const depIds = dependencies.map((x) => x.id) as number[];
+        const actionText = isInstall ? '安装' : '删除';
-      const startTime = dayjs();
+        const depIds = dependencies.map((x) => x.id) as number[];
+        const startTime = dayjs();
 
-      const message = `开始${actionText}依赖 ${depNames}，开始时间 ${startTime.format(
+        const message = `开始${actionText}依赖 ${depNames}，开始时间 ${startTime.format(
-        'YYYY-MM-DD HH:mm:ss',
-      )}\n\n`;
-      this.sockService.sendMessage({
-        type: socketMessageType,
-        message,
-        references: depIds,
-      });
-      await this.updateLog(depIds, message);
-
-      const cp = spawn(`${depRunCommand} ${depNames}`, { shell: '/bin/bash' });
-
-      cp.stdout.on('data', async (data) => {
-        this.sockService.sendMessage({
-          type: socketMessageType,
-          message: data.toString(),
-          references: depIds,
-        });
-        await this.updateLog(depIds, data.toString());
-      });
-
-      cp.stderr.on('data', async (data) => {
-        this.sockService.sendMessage({
-          type: socketMessageType,
-          message: data.toString(),
-          references: depIds,
-        });
-        await this.updateLog(depIds, data.toString());
-      });
-
-      cp.on('error', async (err) => {
-        this.sockService.sendMessage({
-          type: socketMessageType,
-          message: JSON.stringify(err),
-          references: depIds,
-        });
-        await this.updateLog(depIds, JSON.stringify(err));
-        resolve(null);
-      });
-
-      cp.on('close', async (code) => {
-        const endTime = dayjs();
-        const isSucceed = code === 0;
-        const resultText = isSucceed ? '成功' : '失败';
-
-        const message = `\n依赖${actionText}${resultText}，结束时间 ${endTime.format(
           'YYYY-MM-DD HH:mm:ss',
-        )}，耗时 ${endTime.diff(startTime, 'second')} 秒`;
+        )}\n\n`;
         this.sockService.sendMessage({
           type: socketMessageType,
           message,
@@ -230,25 +184,70 @@ export default class DependenceService {
         });
         await this.updateLog(depIds, message);
 
-        let status = null;
+        const cp = spawn(`${depRunCommand} ${depNames}`, { shell: '/bin/bash' });
-        if (isSucceed) {
-          status = isInstall
-            ? DependenceStatus.installed
-            : DependenceStatus.removed;
-        } else {
-          status = isInstall
-            ? DependenceStatus.installFailed
-            : DependenceStatus.removeFailed;
-        }
-        await DependenceModel.update({ status }, { where: { id: depIds } });
 
-        // 如果删除依赖成功或者强制删除
+        cp.stdout.on('data', async (data) => {
-        if ((isSucceed || force) && !isInstall) {
+          this.sockService.sendMessage({
-          this.removeDb(depIds);
+            type: socketMessageType,
-        }
+            message: data.toString(),
+            references: depIds,
+          });
+          await this.updateLog(depIds, data.toString());
+        });
 
-        resolve(null);
+        cp.stderr.on('data', async (data) => {
+          this.sockService.sendMessage({
+            type: socketMessageType,
+            message: data.toString(),
+            references: depIds,
+          });
+          await this.updateLog(depIds, data.toString());
+        });
+
+        cp.on('error', async (err) => {
+          this.sockService.sendMessage({
+            type: socketMessageType,
+            message: JSON.stringify(err),
+            references: depIds,
+          });
+          await this.updateLog(depIds, JSON.stringify(err));
+        });
+
+        cp.on('close', async (code) => {
+          const endTime = dayjs();
+          const isSucceed = code === 0;
+          const resultText = isSucceed ? '成功' : '失败';
+
+          const message = `\n依赖${actionText}${resultText}，结束时间 ${endTime.format(
+            'YYYY-MM-DD HH:mm:ss',
+          )}，耗时 ${endTime.diff(startTime, 'second')} 秒`;
+          this.sockService.sendMessage({
+            type: socketMessageType,
+            message,
+            references: depIds,
+          });
+          await this.updateLog(depIds, message);
+
+          let status = null;
+          if (isSucceed) {
+            status = isInstall
+              ? DependenceStatus.installed
+              : DependenceStatus.removed;
+          } else {
+            status = isInstall
+              ? DependenceStatus.installFailed
+              : DependenceStatus.removeFailed;
+          }
+          await DependenceModel.update({ status }, { where: { id: depIds } });
+
+          // 如果删除依赖成功或者强制删除
+          if ((isSucceed || force) && !isInstall) {
+            this.removeDb(depIds);
+          }
+
+          resolve(null);
+        });
       });
-    });
+    })
   }
 }

back/services/schedule.ts

@@ -9,6 +9,7 @@ import {
   Task,
 } from 'toad-scheduler';
 import dayjs from 'dayjs';
+import { runCronWithLimit } from '../shared/pLimit';
 
 interface ScheduleTaskType {
   id: number;
@@ -40,73 +41,74 @@ export default class ScheduleService {
 
   private maxBuffer = 200 * 1024 * 1024;
 
-  constructor(@Inject('logger') private logger: winston.Logger) {}
+  constructor(@Inject('logger') private logger: winston.Logger) { }
 
   async runTask(
     command: string,
     callbacks: TaskCallbacks = {},
     completionTime: 'start' | 'end' = 'end',
   ) {
-    return new Promise(async (resolve, reject) => {
+    return runCronWithLimit(() => {
-      try {
+      return new Promise(async (resolve, reject) => {
-        const startTime = dayjs();
+        try {
-        await callbacks.onBefore?.(startTime);
+          const startTime = dayjs();
+          await callbacks.onBefore?.(startTime);
 
-        const cp = spawn(command, { shell: '/bin/bash' });
+          const cp = spawn(command, { shell: '/bin/bash' });
 
-        // TODO:
+          callbacks.onStart?.(cp, startTime);
-        callbacks.onStart?.(cp, startTime);
+          completionTime === 'start' && resolve(cp.pid);
-        completionTime === 'start' && resolve(cp.pid);
 
-        cp.stdout.on('data', async (data) => {
+          cp.stdout.on('data', async (data) => {
-          await callbacks.onLog?.(data.toString());
+            await callbacks.onLog?.(data.toString());
-        });
+          });
 
-        cp.stderr.on('data', async (data) => {
+          cp.stderr.on('data', async (data) => {
-          this.logger.info(
+            this.logger.info(
-            '[执行任务失败] %s，时间：%s, 错误信息：%j',
+              '[执行任务失败] %s，时间：%s, 错误信息：%j',
+              command,
+              new Date().toLocaleString(),
+              data.toString(),
+            );
+            await callbacks.onError?.(data.toString());
+          });
+
+          cp.on('error', async (err) => {
+            this.logger.error(
+              '[创建任务失败] %s，时间：%s, 错误信息：%j',
+              command,
+              new Date().toLocaleString(),
+              err,
+            );
+            await callbacks.onError?.(JSON.stringify(err));
+          });
+
+          cp.on('exit', async (code, signal) => {
+            this.logger.info(
+              `[任务退出] ${command} 进程id: ${cp.pid}，退出码 ${code}`,
+            );
+          });
+
+          cp.on('close', async (code) => {
+            const endTime = dayjs();
+            await callbacks.onEnd?.(
+              cp,
+              endTime,
+              endTime.diff(startTime, 'seconds'),
+            );
+            resolve(null);
+          });
+        } catch (error) {
+          await this.logger.error(
+            '执行任务%s失败，时间：%s, 错误信息：%j',
             command,
             new Date().toLocaleString(),
-            data.toString(),
+            error,
           );
-          await callbacks.onError?.(data.toString());
+          await callbacks.onError?.(JSON.stringify(error));
-        });
+        }
-
+      });
-        cp.on('error', async (err) => {
+    })
-          this.logger.error(
-            '[创建任务失败] %s，时间：%s, 错误信息：%j',
-            command,
-            new Date().toLocaleString(),
-            err,
-          );
-          await callbacks.onError?.(JSON.stringify(err));
-        });
-
-        cp.on('exit', async (code, signal) => {
-          this.logger.info(
-            `[任务退出] ${command} 进程id: ${cp.pid}，退出码 ${code}`,
-          );
-        });
-
-        cp.on('close', async (code) => {
-          const endTime = dayjs();
-          await callbacks.onEnd?.(
-            cp,
-            endTime,
-            endTime.diff(startTime, 'seconds'),
-          );
-          resolve(null);
-        });
-      } catch (error) {
-        await this.logger.error(
-          '执行任务%s失败，时间：%s, 错误信息：%j',
-          command,
-          new Date().toLocaleString(),
-          error,
-        );
-        await callbacks.onError?.(JSON.stringify(error));
-      }
-    });
   }
 
   async createCronTask(
@@ -126,12 +128,12 @@ export default class ScheduleService {
     this.scheduleStacks.set(
       _id,
       nodeSchedule.scheduleJob(_id, schedule, async () => {
-        await this.runTask(command, callbacks);
+        this.runTask(command, callbacks);
       }),
     );
 
     if (runImmediately) {
-      await this.runTask(command, callbacks);
+      this.runTask(command, callbacks);
     }
   }
 

back/services/subscription.ts

@@ -8,9 +8,6 @@ import {
 } from '../data/subscription';
 import {
   ChildProcessWithoutNullStreams,
-  exec,
-  execSync,
-  spawn,
 } from 'child_process';
 import fs from 'fs';
 import {
@@ -20,6 +17,7 @@ import {
   createFile,
   killTask,
   handleLogPath,
+  promiseExec,
 } from '../config/util';
 import { promises, existsSync } from 'fs';
 import { FindOptions, Op } from 'sequelize';
@@ -39,7 +37,7 @@ export default class SubscriptionService {
     private scheduleService: ScheduleService,
     private sockService: SockService,
     private sshKeyService: SshKeyService,
-  ) {}
+  ) { }
 
   public async list(searchText?: string): Promise<Subscription[]> {
     let query = {};
@@ -110,18 +108,6 @@ export default class SubscriptionService {
     this.sshKeyService.setSshConfig(docs);
   }
 
-  private async promiseExec(command: string): Promise<string> {
-    return new Promise((resolve, reject) => {
-      exec(
-        command,
-        { maxBuffer: 200 * 1024 * 1024, encoding: 'utf8' },
-        (err, stdout, stderr) => {
-          resolve(stdout || stderr || JSON.stringify(err));
-        },
-      );
-    });
-  }
-
   private taskCallbacks(doc: Subscription): TaskCallbacks {
     return {
       onBefore: async (startTime) => {
@@ -144,7 +130,7 @@ export default class SubscriptionService {
         try {
           if (doc.sub_before) {
             fs.appendFileSync(absolutePath, `\n## 执行before命令...\n\n`);
-            beforeStr = await this.promiseExec(doc.sub_before);
+            beforeStr = await promiseExec(doc.sub_before);
           }
         } catch (error: any) {
           beforeStr =
@@ -171,7 +157,7 @@ export default class SubscriptionService {
         try {
           if (sub.sub_after) {
             fs.appendFileSync(absolutePath, `\n\n## 执行after命令...\n\n`);
-            afterStr = await this.promiseExec(sub.sub_after);
+            afterStr = await promiseExec(sub.sub_after);
           }
         } catch (error: any) {
           afterStr =
@@ -290,10 +276,9 @@ export default class SubscriptionService {
       { status: SubscriptionStatus.queued },
       { where: { id: ids } },
     );
-    concurrentRun(
+    ids.forEach(id => {
-      ids.map((id) => async () => await this.runSingle(id)),
+      this.runSingle(id)
-      10,
+    })
-    );
   }
 
   public async stop(ids: number[]) {
@@ -330,7 +315,7 @@ export default class SubscriptionService {
 
     const command = formatCommand(subscription);
 
-    await this.scheduleService.runTask(
+    this.scheduleService.runTask(
       command,
       this.taskCallbacks(subscription),
     );

back/services/user.ts

@@ -15,7 +15,6 @@ import { NotificationInfo } from '../data/notify';
 import NotificationService from './notify';
 import { Request } from 'express';
 import ScheduleService from './schedule';
-import { spawn } from 'child_process';
 import SockService from './sock';
 import dayjs from 'dayjs';
 

back/shared/pLimit.ts

@@ -0,0 +1,10 @@
+import pLimit from "p-limit";
+import os from 'os';
+
+const cronLimit = pLimit(os.cpus.length);
+
+export function runCronWithLimit<T>(fn: () => Promise<T>): Promise<T> {
+  return cronLimit(() => {
+    return fn();
+  });
+}

back/shared/runCron.ts

@@ -0,0 +1,37 @@
+import { spawn } from "child_process";
+import { runCronWithLimit } from "./pLimit";
+import Logger from '../loaders/logger';
+
+export function runCron(cmd: string): Promise<number> {
+  return runCronWithLimit(() => {
+    return new Promise(async (resolve: any) => {
+      Logger.silly('运行命令: ' + cmd);
+
+      const cp = spawn(cmd, { shell: '/bin/bash' });
+
+      cp.stderr.on('data', (data) => {
+        Logger.info(
+          '[执行任务失败] %s，时间：%s, 错误信息：%j',
+          cmd,
+          new Date().toLocaleString(),
+          data.toString(),
+        );
+      });
+      cp.on('error', (err) => {
+        Logger.error(
+          '[创建任务失败] %s，时间：%s, 错误信息：%j',
+          cmd,
+          new Date().toLocaleString(),
+          err,
+        );
+      });
+
+      cp.on('close', async (code) => {
+        Logger.info(
+          `[任务退出] ${cmd} 进程id: ${cp.pid} 退出，退出码 ${code}`,
+        );
+        resolve();
+      });
+    });
+  })
+}

package.json

@@ -81,6 +81,7 @@
     "nedb": "^1.8.0",
     "node-schedule": "^2.1.0",
     "nodemailer": "^6.7.2",
+    "p-limit": "^4.0.0",
     "protobufjs": "^7.2.3",
     "pstree.remy": "^1.1.8",
     "reflect-metadata": "^0.1.13",

pnpm-lock.yaml

@@ -79,6 +79,9 @@ dependencies:
   nodemailer:
     specifier: ^6.7.2
     version: 6.9.1
+  p-limit:
+    specifier: ^4.0.0
+    version: 4.0.0
   protobufjs:
     specifier: ^7.2.3
     version: 7.2.3
@@ -10844,6 +10847,13 @@ packages:
       yocto-queue: 0.1.0
     dev: true
 
+  /p-limit@4.0.0:
+    resolution: {integrity: sha512-5b0R4txpzjPWVw/cXXUResoD4hb6U/x9BH08L7nw+GN1sezDzPdxeRvpc9c433fZhBan/wusjbCsqwqm4EIBIQ==}
+    engines: {node: ^12.20.0 || ^14.13.1 || >=16.0.0}
+    dependencies:
+      yocto-queue: 1.0.0
+    dev: false
+
   /p-locate@4.1.0:
     resolution: {integrity: sha512-R79ZZ/0wAxKGu3oYMlz8jy/kbhsNrS7SKZ7PxEHBgJ5+F2mtFW2fK2cOtBh1cHYkQsbzFV7I+EoRKe6Yt0oK7A==}
     engines: {node: '>=8'}
@@ -15471,6 +15481,11 @@ packages:
     engines: {node: '>=10'}
     dev: true
 
+  /yocto-queue@1.0.0:
+    resolution: {integrity: sha512-9bnSc/HEW2uRy67wc+T8UwauLuPJVn28jb+GtJY16iiKWyvmYJRXVT4UamsAEGQfPohgr2q4Tq0sQbQlxTfi1g==}
+    engines: {node: '>=12.20'}
+    dev: false
+
   /yorkie@2.0.0:
     resolution: {integrity: sha512-jcKpkthap6x63MB4TxwCyuIGkV0oYP/YRyuQU5UO0Yz/E/ZAu+653/uov+phdmO54n6BcvFRyyt0RRrWdN2mpw==}
     engines: {node: '>=4'}