Cursor rule

1---
2description: 任务与队列(Task)
3globs: 
4---
5# 任务与队列(Task)
6 
7## 内置任务（代码中配置）
8 
9内置定时任务能力来自于[midwayjs](https://www.midwayjs.org/docs/extensions/cron)
10 
11### 引入组件
12 
13```ts
14import { Configuration } from "@midwayjs/core";
15import * as cron from "@midwayjs/cron"; // 导入模块
16import { join } from "path";
17 
18@Configuration({
19  imports: [cron],
20  importConfigs: [join(__dirname, "config")],
21})
22export class AutoConfiguration {}
23```
24 
25### 使用
26 
27```ts
28import { Job, IJob } from "@midwayjs/cron";
29import { FORMAT } from "@midwayjs/core";
30 
31@Job({
32  cronTime: FORMAT.CRONTAB.EVERY_PER_30_MINUTE,
33  start: true,
34})
35export class DataSyncCheckerJob implements IJob {
36  async onTick() {
37    // ...
38  }
39}
40```
41 
42```ts
43@Job("syncJob", {
44  cronTime: "*/2 * * * * *", // 每隔 2s 执行
45})
46export class DataSyncCheckerJob implements IJob {
47  async onTick() {
48    // ...
49  }
50}
51```
52 
53### 规则 cron
54 
55```ts
56*    *    *    *    *    *
57┬    ┬    ┬    ┬    ┬    ┬
58│    │    │    │    │    |
59│    │    │    │    │    └ day of week (0 - 7) (0 or 7 is Sun)
60│    │    │    │    └───── month (1 - 12)
61│    │    │    └────────── day of month (1 - 31)
62│    │    └─────────────── hour (0 - 23)
63│    └──────────────────── minute (0 - 59)
64└───────────────────────── second (0 - 59, optional)
65 
66```
67 
68::: warning 警告
69 
70注意：该方式在多实例部署的情况下无法做到任务之前的协同，任务存在重复执行的可能
71 
72:::
73 
74## 本地任务（管理后台配置，v8.0 新增）
75 
76可以到登录后台`/系统管理/任务管理/任务列表`，配置任务。默认是不需要任何依赖的， 旧版需要依赖`redis`才能使用该功能。
77 
78### 配置任务
79 
80配置完任务可以调用你配置的 service 方法，如：taskDemoService.test()
81 
82### 规则 cron
83 
84规则 cron
85 
86```ts
87*    *    *    *    *    *
88┬    ┬    ┬    ┬    ┬    ┬
89│    │    │    │    │    |
90│    │    │    │    │    └ day of week (0 - 7) (0 or 7 is Sun)
91│    │    │    │    └───── month (1 - 12)
92│    │    │    └────────── day of month (1 - 31)
93│    │    └─────────────── hour (0 - 23)
94│    └──────────────────── minute (0 - 59)
95└───────────────────────── second (0 - 59, optional)
96 
97```
98 
99规则示例：
100 
101- 每 5 秒执行一次: `*/5 * * * * *`
102- 每 5 分钟执行一次: `*/5 * * * *`
103- 每小时执行一次: `0 * * * *`
104- 每天执行一次: `0 0 * * *`
105- 每天 1 点执行: `0 1 * * *`
106- 每周执行一次: `0 0 * * 0`
107- 每月执行一次: `0 0 1 * *`
108 
109![](/admin/node/task.png)
110 
111## 分布式任务（管理后台配置）
112 
113当需要分布式部署时，需要开启分布式任务，通过 redis 作为协同整个集群的任务，防止任务重复执行等异常情况。
114 
115#### 引入插件
116 
117`src/configuration.ts`
118 
119```ts
120import { Configuration, App } from "@midwayjs/core";
121import { join } from "path";
122import * as task from "@cool-midway/task";
123 
124@Configuration({
125  imports: [task],
126  importConfigs: [join(__dirname, "./config")],
127})
128export class ContainerLifeCycle {
129  @App()
130  app: koa.Application;
131 
132  async onReady() {}
133}
134```
135 
136#### 配置
137 
138[redis>=5.x](https://redis.io/)，推荐[redis>=7.x](https://redis.io/)
139 
140`src/config/config.default.ts`
141 
142::: warning 注意
143很多人忽略了这个配置，导致项目包 redis 连接错误！！！
144:::
145 
146```ts
147import { CoolFileConfig, MODETYPE } from "@cool-midway/file";
148import { MidwayConfig } from "@midwayjs/core";
149import * as fsStore from "cache-manager-fs-hash";
150 
151export default {
152  // 修改成你自己独有的key
153  keys: "cool-admin for node",
154  koa: {
155    port: 8001,
156  },
157  // cool配置
158  cool: {
159    redis: {
160      host: "127.0.0.1",
161      port: 6379,
162      password: "",
163      db: 0,
164    },
165  },
166} as unknown as MidwayConfig;
167```
168 
169redis cluster 方式
170 
171```ts
172[
173  {
174    host: "192.168.0.103",
175    port: 7000,
176  },
177  {
178    host: "192.168.0.103",
179    port: 7001,
180  },
181  {
182    host: "192.168.0.103",
183    port: 7002,
184  },
185  {
186    host: "192.168.0.103",
187    port: 7003,
188  },
189  {
190    host: "192.168.0.103",
191    port: 7004,
192  },
193  {
194    host: "192.168.0.103",
195    port: 7005,
196  },
197];
198```
199 
200### 创建执行任务的 service
201 
202```ts
203import { Provide } from "@midwayjs/core";
204import { BaseService } from "@cool-midway/core";
205/**
206 * 任务执行的demo示例
207 */
208@Provide()
209export class DemoTaskService extends BaseService {
210  /**
211   * 测试任务执行
212   * @param params 接收的参数 数组 [] 可不传
213   */
214  async test(params?: []) {
215    // 需要登录后台任务管理配置任务
216    console.log("任务执行了", params);
217  }
218}
219```
220 
221### 配置定时任务
222 
223登录后台 任务管理/任务列表
224 
225![](/admin/node/task.png)
226 
227::: warning
228截图中的 demoTaskService 为上一步执行任务的 service 的实例 ID，midwayjs 默认为类名首字母小写！！！
229 
230任务调度基于 redis，所有的任务都需要通过代码去维护任务的创建，启动，暂停。 所以直接改变数据库的任务状态是无效的，redis 中的信息还未清空， 任务将继续执行。
231:::
232 
233## 队列
234 
235之前的分布式任务调度，其实是利用了[bullmq](https://docs.bullmq.io/)的重复队列机制。
236 
237在项目开发过程中特别是较大型、数据量较大、业务较复杂的场景下往往需要用到队列。 如：抢购、批量发送消息、分布式事务、订单 2 小时后失效等。
238 
239得益于[bullmq](https://docs.bullmq.io/)，cool 的队列也支持`延迟`、`重复`、`优先级`等高级特性。
240 
241### 创建队列
242 
243一般放在名称为 queue 文件夹下
244 
245#### 普通队列
246 
247普通队列数据由消费者自动消费，必须重写 data 方法用于被动消费数据。
248 
249`src/modules/demo/queue/comm.ts`
250 
251```ts
252import { BaseCoolQueue, CoolQueue } from "@cool-midway/task";
253import { IMidwayApplication } from "@midwayjs/core";
254import { App } from "@midwayjs/core";
255 
256/**
257 * 普通队列
258 */
259@CoolQueue()
260export class DemoCommQueue extends BaseCoolQueue {
261  @App()
262  app: IMidwayApplication;
263 
264  async data(job: any, done: any): Promise<void> {
265    // 这边可以执行定时任务具体的业务或队列的业务
266    console.log("数据", job.data);
267    // 抛出错误 可以让队列重试，默认重试5次
268    //throw new Error('错误');
269    done();
270  }
271}
272```
273 
274#### 主动队列
275 
276主动队列数据由消费者主动消费
277 
278`src/modules/demo/queue/getter.ts`
279 
280```ts
281import { BaseCoolQueue, CoolQueue } from "@cool-midway/task";
282 
283/**
284 * 主动消费队列
285 */
286@CoolQueue({ type: "getter" })
287export class DemoGetterQueue extends BaseCoolQueue {}
288```
289 
290主动消费数据
291 
292```ts
293 // 主动消费队列
294  @Inject()
295  demoGetterQueue: DemoGetterQueue;
296 
297  const job = await this.demoGetterQueue.getters.getJobs(['wait'], 0, 0, true);
298  // 获得完将数据从队列移除
299  await job[0].remove();
300```
301 
302### 发送数据
303 
304```ts
305import { Get, Inject, Post, Provide } from "@midwayjs/core";
306import { CoolController, BaseController } from "@cool-midway/core";
307import { DemoCommQueue } from "../../queue/comm";
308import { DemoGetterQueue } from "../../queue/getter";
309 
310/**
311 * 队列
312 */
313@Provide()
314@CoolController()
315export class DemoQueueController extends BaseController {
316  // 普通队列
317  @Inject()
318  demoCommQueue: DemoCommQueue;
319 
320  // 主动消费队列
321  @Inject()
322  demoGetterQueue: DemoGetterQueue;
323 
324  /**
325   * 发送数据到队列
326   */
327  @Post("/add", { summary: "发送队列数据" })
328  async queue() {
329    this.demoCommQueue.add({ a: 2 });
330    return this.ok();
331  }
332 
333  /**
334   * 获得队列中的数据，只有当队列类型为getter时有效
335   */
336  @Get("/getter")
337  async getter() {
338    const job = await this.demoCommQueue.getters.getJobs(["wait"], 0, 0, true);
339    // 获得完将数据从队列移除
340    await job[0].remove();
341    return this.ok(job[0].data);
342  }
343}
344```
345 
346队列配置
347 
348```ts
349interface JobOpts {
350  priority: number; // Optional priority value. ranges from 1 (highest priority) to MAX_INT  (lowest priority). Note that
351  // using priorities has a slight impact on performance, so do not use it if not required.
352 
353  delay: number; // An amount of milliseconds to wait until this job can be processed. Note that for accurate delays, both
354  // server and clients should have their clocks synchronized. [optional].
355 
356  attempts: number; // The total number of attempts to try the job until it completes.
357 
358  repeat: RepeatOpts; // Repeat job according to a cron specification.
359 
360  backoff: number | BackoffOpts; // Backoff setting for automatic retries if the job fails, default strategy: `fixed`
361 
362  lifo: boolean; // if true, adds the job to the right of the queue instead of the left (default false)
363  timeout: number; // The number of milliseconds after which the job should be fail with a timeout error [optional]
364 
365  jobId: number | string; // Override the job ID - by default, the job ID is a unique
366  // integer, but you can use this setting to override it.
367  // If you use this option, it is up to you to ensure the
368  // jobId is unique. If you attempt to add a job with an id that
369  // already exists, it will not be added.
370 
371  removeOnComplete: boolean | number; // If true, removes the job when it successfully
372  // completes. A number specified the amount of jobs to keep. Default behavior is to keep the job in the completed set.
373 
374  removeOnFail: boolean | number; // If true, removes the job when it fails after all attempts. A number specified the amount of jobs to keep
375  // Default behavior is to keep the job in the failed set.
376  stackTraceLimit: number; // Limits the amount of stack trace lines that will be recorded in the stacktrace.
377}
378```
379 
380::: tip
381this.demoQueue.queue 获得的就是 bull 实例，更多 bull 的高级用户可以查看[bull 文档](https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md)
382:::
383