[Question] 任务队列 (bull) runJob 分布式的情况下出现并发执行，是更新导致的？

[7kyun]

Describe the problem（描述问题）

如题

之前 3.8.0 表现是正常的，及时并发运行 queue.runJob() 也是队列一个个执行
前几天升级为 当前版本 3.16.0 出现上述问题

部分示例代码

// Controller
const queue = this.bullFramework.getQueue('processor');
await queue.runJob();

// Processor
import { Context, IProcessor, Processor } from '@midwayjs/bull';
import { Inject } from '@midwayjs/core';

@Processor('processor', {
  repeat: {
    // cron: '0 0/5 * * * *', // 每5分钟执行一次
    cron: '5 * * * * *', // 每5分钟执行一次
  },
  removeOnComplete: 3, // 成功后移除任务记录，最多保留最近 3 条记录
  removeOnFail: 10, // 失败后移除任务记录，最多保留最近 10 条记录
})
export class OrderRepairProcessor implements IProcessor {
  @Inject()
  ctx: Context;

  async execute(options = { isAuto: true }) {
    const looper = (ms = 1000) => {
      return new Promise(resolve => {
        setTimeout(() => {
          this.ctx.logger.info('[processor - looper] 执行 %s', new Date());
          resolve(true);
        }, ms);
      });
    };
    this.ctx.logger.info('[processor] 开始 %s', new Date());
    await looper(10000);
    this.ctx.logger.info('[processor] 结束 %s', new Date());
  }
}

表现

Midway Versions（Midway 版本）

>> Start to check your midway component version...

✓ @midwayjs/faas-typings(not installed)
✓ @midwayjs/fc-starter(not installed)
✓ @midwayjs/serverless-http-parser(not installed)
✓ @midwayjs/async-hooks-context-manager(3.16.2)
✓ @midwayjs/axios(not installed)
✓ @midwayjs/bootstrap(3.16.2)
✓ @midwayjs/bull(3.16.5)
✓ @midwayjs/bull-board(3.16.5)
✓ @midwayjs/cache-manager(not installed)
✓ @midwayjs/captcha(not installed)
✓ @midwayjs/casbin(not installed)
✓ @midwayjs/casbin-redis-adapter(not installed)
✓ @midwayjs/casbin-typeorm-adapter(not installed)
✓ @midwayjs/code-dye(not installed)
✓ @midwayjs/consul(not installed)
✓ @midwayjs/core(3.16.2)
✓ @midwayjs/cos(not installed)
✓ @midwayjs/cron(not installed)
✓ @midwayjs/cross-domain(not installed)
✓ @midwayjs/decorator(3.16.2)
✓ @midwayjs/etcd(not installed)
✓ @midwayjs/express-session(not installed)
✓ @midwayjs/faas(not installed)
✓ @midwayjs/grpc(not installed)
✓ @midwayjs/http-proxy(not installed)
✓ @midwayjs/i18n(3.16.5)
✓ @midwayjs/info(3.16.5)
✓ @midwayjs/jwt(3.16.5)
✓ @midwayjs/kafka(not installed)
✓ @midwayjs/leoric(not installed)
✓ @midwayjs/mikro(not installed)
✓ @midwayjs/mock(3.16.5)
✓ @midwayjs/mongoose(not installed)
✓ @midwayjs/mqtt(not installed)
✓ @midwayjs/oss(3.16.5)
✓ @midwayjs/otel(not installed)
✓ @midwayjs/passport(not installed)
✓ @midwayjs/process-agent(not installed)
✓ @midwayjs/prometheus(not installed)
✓ @midwayjs/prometheus-socket-io(not installed)
✓ @midwayjs/rabbitmq(not installed)
✓ @midwayjs/redis(3.16.5)
✓ @midwayjs/security(not installed)
✓ @midwayjs/sequelize(not installed)
✓ @midwayjs/session(3.16.5)
✓ @midwayjs/socketio(not installed)
✓ @midwayjs/static-file(not installed)
✓ @midwayjs/swagger(3.16.5)
✓ @midwayjs/tablestore(not installed)
✓ @midwayjs/tags(not installed)
✓ @midwayjs/tenant(not installed)
✓ @midwayjs/typegoose(not installed)
✓ @midwayjs/typeorm(3.16.5)
✓ @midwayjs/upload(3.16.5)
✓ @midwayjs/validate(3.16.5)
✓ @midwayjs/version(3.16.5)
✓ @midwayjs/view(not installed)
✓ @midwayjs/view-ejs(not installed)
✓ @midwayjs/view-nunjucks(not installed)
✓ @midwayjs/web(not installed)
✓ @midwayjs/express(not installed)
✓ @midwayjs/koa(3.16.5)
✓ @midwayjs/ws(not installed)
**********************************************************************
>> Check complete, all versions are healthy.
**********************************************************************

[czy88840616]

队列执行的个数取决于你有几个 worker 连接到 redis 去执行 job

[7kyun]

如果是分布式的环境，存在两个容器，那就会并发两个去执行？
之前不是没有配置 Concurrency 即使是多个worker也是一个个出队的吗
3.8.0升级到3.16.0之后才会这样，这段逻辑之前已经跑了一年多好像都是按预期执行的

[czy88840616]

Concurrency 在当前应该没什么用，都是 1，我理解实际执行是按连到 redis 上的 client 来算的。

[czy88840616]

我扒了下 src 下面的变化，2022.11.18号之后，粗略看了下没有特别相关的。我没有环境，不太看得出哪个变更有影响。

[7kyun]

所以这玩意其实一直都是这样运行的？
只要到了执行时间，有注册了并空闲的服务都会去执行？🥲

[czy88840616]

试了下，我简单描述下，测试代码是并发 2，定时 1s，任务执行会等待 10s，启动了两个子进程。

let count = 0;

@Processor('test', 2, {
  repeat: {
    cron: FORMAT.CRONTAB.EVERY_SECOND
  },
})
export class TestProcessor implements IProcessor {
  async execute(data) {
    console.log(`${new Date().toLocaleString()} 进程：${process.pid} processor start: ${count++}`);
    await new Promise(resolve => setTimeout(resolve, 10000));
    console.log(`${new Date().toLocaleString()} 进程：${process.pid} processor end`);
  }
}

结果如下：

每隔 5s，会定时触发一次任务，两个进程号为 8742 和 8743。

• 1、38s 时，8742 触发了任务，编号 0
• 2、39s 时，8742 触发了任务，编号 1
• 3、40s 时，8743 触发了任务，编号 0
• 4、41s 时，8743 触发了任务，编号 1
• 5、48s 时，8742 编号 0 任务执行完毕

说明每个 worker 的并发度是独立计算的，并发的配置指的是每个 worker 最多可同时执行 job 的数量，而不是同时服务端下发的任务数量。

[7kyun]

辛苦大佬 🥺
确实是这样的，后面我从 nestjs 中有看到这方面的介绍

每个worker空闲时可处理的 job 数量取决于当前worker的 concurrency 值

我之前的理解是: 所有worker共享concurrency，如果为1，则job是一个个出队
这是错误的

最后再次感谢大佬的解答

[waitingsong]

那如果要实现 “所有worker共享concurrency”，是不是得整个消息队列呢？

[czy88840616]

如果消息数量大于 worker 数量，那 concurrency 设为1，自然每个 worker 只会执行一条了。

如果需要严格保证唯一，那就用动态 api，每个 worker 执行完，再添加一个任务进队列就行了吧。

[7kyun]

是的，前几天临时就这么处理的，先用着吧

…

---- 回复的原邮件 ---- | 发件人 | Harry ***@***.***> | | 发送日期 | 2024年08月29日 22:56 | | 收件人 | midwayjs/midway ***@***.***> | | 抄送人 | Kyun ***@***.***>, Author ***@***.***> | | 主题 | Re: [midwayjs/midway] [Question] 任务队列 (bull) runJob 分布式的情况下出现并发执行，是更新导致的？ (Discussion #4026) | 如果消息数量大于 worker 数量，那 concurrency 设为1，自然每个 worker 只会执行一条了。 如果需要严格保证唯一，那就用动态 api，每个 worker 执行完，再添加一个任务进队列就行了吧。 — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you authored the thread.Message ID: ***@***.***>