完善分布式任务调度

src/modules/task/middleware/task.ts

@@ -0,0 +1,31 @@
+import { CoolCommException } from '@cool-midway/core';
+import { Inject, Middleware } from '@midwayjs/decorator';
+import { NextFunction, Context } from '@midwayjs/koa';
+import { IMiddleware } from '@midwayjs/core';
+import { TaskInfoQueue } from '../queue/task';
+
+/**
+ * 任务中间件
+ */
+@Middleware()
+export class TaskMiddleware implements IMiddleware<Context, NextFunction> {
+  @Inject()
+  taskInfoQueue: TaskInfoQueue;
+  resolve() {
+    return async (ctx: Context, next: NextFunction) => {
+      const urls = ctx.url.split('/');
+      if (
+        ['add', 'update', 'once', 'stop', 'start'].includes(
+          urls[urls.length - 1]
+        )
+      ) {
+        if (!this.taskInfoQueue.metaQueue) {
+          throw new CoolCommException(
+            'task插件未启用或redis配置错误或redis版本过低(>=6.x)'
+          );
+        }
+      }
+      await next();
+    };
+  }
+}

src/modules/task/service/info.ts

@@ -294,13 +294,13 @@ export class TaskInfoService extends BaseService {
     const task = await this.taskInfoEntity.findOne({ id: job.id });
     const nextTime = await this.getNextRunTime(task.id);
     if (task) {
-      if (task.nextRunTime.getTime() == nextTime.getTime()) {
+      //   if (task.nextRunTime.getTime() == nextTime.getTime()) {
-        task.status = 0;
+      //     task.status = 0;
-        task.nextRunTime = nextTime;
+      //     task.nextRunTime = nextTime;
-        this.taskInfoQueue.removeRepeatableByKey(job.key);
+      //     this.taskInfoQueue.removeRepeatableByKey(job.key);
-      } else {
+      //   } else {
-        task.nextRunTime = nextTime;
+      task.nextRunTime = nextTime;
-      }
+      //   }
       await this.taskInfoEntity.update(task.id, task);
     }
   }