[Feature request] 队列组件是否可以支持，指定队列进程数量

[yuntian001]

The feature, motivation（功能、动机）

例如，我用pm2多进程部署，假设开启进程数为6，我希望4个进程仅启用koa组件监听端口，2个进程仅处理队列任务，以防止队列任务运行时阻塞http请求或者http请求阻塞队列任务；但是，koa组件还可以需要允许动态投递队列任务。是否可以对应配置。

Alternatives（替代方案）

No response

Additional context（其他上下文）

No response

[czy88840616]

可以，这个场景我理解下来不一定需要在 bullmq 组件里新增“指定队列进程数量”的配置，更适合用不同启动入口来区分进程角色。

BullMQ 本身是基于 Redis 的分布式队列，producer 和 consumer 不要求在同一个 Node 进程里。也就是说，同一套代码可以拆两个 Midway 入口：

• api 入口：启动 Koa + bullmq，只负责 HTTP 和投递任务；
• worker 入口：只启动 bullmq，加载 @Processor，只消费任务。

一个简化例子如下。

// src/common/task.service.ts
import { Inject, Provide } from '@midwayjs/core';
import * as bullmq from '@midwayjs/bullmq';

@Provide()
export class TaskService {
  @Inject()
  bullmqFramework: bullmq.Framework;

  async submit(data: any) {
    const queue = this.bullmqFramework.getQueue('image-task')
      ?? this.bullmqFramework.createQueue('image-task');

    return queue.addJobToQueue(data);
  }
}

// src/processor/image.processor.ts
import { Processor, IProcessor } from '@midwayjs/bullmq';

@Processor('image-task', {}, { concurrency: 2 })
export class ImageProcessor implements IProcessor {
  async execute(data: any) {
    // 这里只在 worker 入口加载时消费任务
  }
}

// src/api/configuration.ts
import { Configuration } from '@midwayjs/core';
import * as koa from '@midwayjs/koa';
import * as bullmq from '@midwayjs/bullmq';

@Configuration({
  imports: [koa, bullmq],
  importConfigs: ['./config'],
})
export class ApiConfiguration {}

// src/worker/configuration.ts
import { Configuration } from '@midwayjs/core';
import * as bullmq from '@midwayjs/bullmq';
import '../processor/image.processor';

@Configuration({
  imports: [bullmq],
  importConfigs: ['./config'],
})
export class WorkerConfiguration {}

然后用两个 bootstrap 指到不同入口目录，例如：

// bootstrap-api.js
const { join } = require('path');
const { Bootstrap } = require('@midwayjs/bootstrap');

Bootstrap.configure({
  baseDir: join(__dirname, 'dist/api'),
}).run();

// bootstrap-worker.js
const { join } = require('path');
const { Bootstrap } = require('@midwayjs/bootstrap');

Bootstrap.configure({
  baseDir: join(__dirname, 'dist/worker'),
}).run();

PM2 里可以配两组 app：

module.exports = {
  apps: [
    { name: 'api', script: './bootstrap-api.js', instances: 4 },
    { name: 'queue-worker', script: './bootstrap-worker.js', instances: 2 },
  ],
};

这样 api 进程因为没有加载 processor 文件，不会创建 BullMQ Worker，但仍然可以通过 createQueue/getQueue 投递任务；queue-worker 进程只加载 bullmq 和 processor，主框架就是 bullmq，只负责消费。业务 service、DTO、配置都可以复用同一套代码。

如果后续要补充能力，我觉得更适合补文档/示例，说明如何用多入口拆分 producer/consumer；组件层面未必需要按 PM2 进程数再做一层分配逻辑。

[czy88840616]

ESM 项目不要用 CommonJSFileDetector，可以换成 ESModuleFileDetector，ignore 这些参数是一样的。

import { Configuration, ESModuleFileDetector } from '@midwayjs/core';
import * as koa from '@midwayjs/koa';
import * as bullmq from '@midwayjs/bullmq';

@Configuration({
  imports: [koa, bullmq],
  detector: new ESModuleFileDetector({
    ignore: ['**/processor/**'],
  }),
  importConfigs: ['./config'],
})
export class ApiConfiguration {}

这样思路还是一样：API 进程导入 bullmq 但不扫描 processor，worker 进程再扫描 processor。

后续加“是否创建 worker”的配置我觉得可以考虑，可能会设计成类似 bullmq.worker.enable = false / bullmq.processor.enable = false 这种 producer-only 开关，这样对多入口场景会更直观。

[RomenMark]

能否再给出一个详细的示例？

 baseDir: join(__dirname, 'dist/worker'),
配置baseDir，原先controller等文件夹路径是否就不对了

[czy88840616]

是的，如果把 API 入口的 baseDir 指到 dist/worker，默认扫描根目录就会变成 dist/worker，原来的 controller/service 路径当然会不对。这里我前面的示例写得不够严谨。

更推荐这样理解：baseDir 是当前进程的应用入口/默认扫描根目录；要拆 producer / consumer，关键是控制“这个入口扫描哪些目录”，不一定要把 baseDir 改到一个只包含 worker 的目录。

常见两种写法：

1. API 进程仍然用正常的 dist 作为 baseDir，然后在 API 的 detector 里忽略 processor：

import { Configuration, ESModuleFileDetector } from '@midwayjs/core';
import * as koa from '@midwayjs/koa';
import * as bullmq from '@midwayjs/bullmq';

@Configuration({
  imports: [koa, bullmq],
  detector: new ESModuleFileDetector({
    ignore: ['**/processor/**'],
  }),
  importConfigs: ['./config'],
})
export class ApiConfiguration {}

2. 如果你确实拆成 dist/api、dist/worker 这种目录，就不要依赖默认扫描目录，而是在 detector 里显式写 loadDir，例如 API 入口扫描 controller/service，但不扫描 processor；worker 入口只扫描 processor 和它依赖的公共 service。类似：

import { join } from 'node:path';
import { Configuration, ESModuleFileDetector } from '@midwayjs/core';

@Configuration({
  detector: new ESModuleFileDetector({
    loadDir: [
      join(__dirname, '../controller'),
      join(__dirname, '../service'),
      join(__dirname, '../middleware'),
    ],
    ignore: ['**/processor/**'],
  }),
})
export class ApiConfiguration {}

所以你的判断是对的：baseDir: dist/worker 只适合 worker 入口自己使用；API 入口不要这样配。API 入口要么保持 baseDir = dist 并 ignore processor，要么在拆目录后用 loadDir 明确把 controller/service 加回来。

[RomenMark]

感谢回复，这样确实将producer / consumer拆分开来了。还有种场景疑问，如果要继续拆分，定时任务（重复执行的任务）和普通任务，有没有比较好的方案。当多线程运行repeat的任务时，会重复运行。是否继续加一个单独的入口用于repeat的任务，用单线程启动？

[yuntian001]

ESM 项目不要用 CommonJSFileDetector，可以换成 ESModuleFileDetector，ignore 这些参数是一样的。

import { Configuration, ESModuleFileDetector } from '@midwayjs/core';
import * as koa from '@midwayjs/koa';
import * as bullmq from '@midwayjs/bullmq';

@Configuration({
  imports: [koa, bullmq],
  detector: new ESModuleFileDetector({
    ignore: ['**/processor/**'],
  }),
  importConfigs: ['./config'],
})
export class ApiConfiguration {}

这样思路还是一样：API 进程导入 bullmq 但不扫描 processor，worker 进程再扫描 processor。

后续加“是否创建 worker”的配置我觉得可以考虑，可能会设计成类似 bullmq.worker.enable = false / bullmq.processor.enable = false 这种 producer-only 开关，这样对多入口场景会更直观。

producer-only 开关，什么时间能加上，现在是用的忽略加载processor文件的方式做的。但是当只需要投递的进程忽略加载processor文件时，会连queue都获取不到，需要手动再创建一遍才行。手动创建的话queue参数就不是在Processor 获取到的queueOptions 了很容易导致不一致。

[czy88840616]

这个场景建议再拆一层来看：repeat job 的“注册/调度”和 Worker 的“消费”不是一回事。

BullMQ 里同一个队列可以有多个 Worker，单个 job 正常只会被其中一个 Worker 拿到执行，所以普通任务不需要担心多个 worker 同时执行同一条 job。但 Midway 现在通过 @Processor(..., { repeat }) 注册重复任务时，是在 bullmq 框架启动时做的；如果多个入口都扫描到这个 repeat processor，就等于多个进程都参与 repeat 注册，尤其当前还有 clearRepeatJobWhenStart 默认清理旧 repeat/job scheduler 的逻辑，多进程启动时不太适合都做这件事。

所以更稳的拆法是：

1. 单独一个 repeat-scheduler 入口，只扫描带 repeat 配置的 processor，PM2 instances: 1，负责注册/维护重复任务；
2. 普通 worker 入口可以多实例，只负责消费队列；
3. 如果 repeat 任务本身也要求严格串行执行，可以把 repeat 任务放到独立 queue，worker 的 concurrency 设为 1，并且这个 repeat queue 只启动 1 个 worker 进程。

也就是说，不一定所有 repeat 任务都要单线程入口：如果只是避免重复触发，关键是 repeat 注册入口单实例；如果业务要求同一类 repeat 任务绝对串行，再把对应 queue 的消费也限制为单 worker / concurrency 1。后续如果加 producer-only 或 worker-enable 开关，也应该把这类 repeat 注册场景一起考虑进去。

[RomenMark]

这里还发现一个现象，如果想共用config，typeorm中的配置项如果用目录扫描的方式,会报错找不到metadata。只能是 entities: [ActivityEntity]，引入每一个entity

[czy88840616]

可以写个脚本去扫描，不走内置的扫描方式

[czy88840616]

这个现象大概率还是和多入口后的 baseDir / 加载路径有关。

Midway 里的 TypeORM entities 字符串配置不是直接交给 TypeORM 原生 glob 处理的，框架会先按当前进程的 baseDir 去扫描并加载成 entity class，再传给 TypeORM。所以如果 worker/API 入口把 baseDir 指到了局部目录，比如 dist/worker，那么 entities: ['entity']、entities: ['**/*.entity.{j,t}s'] 这类相对路径就只会按这个局部目录去找，公共 entity 不在这个扫描根下时就容易出现 entity 没加载到，或者 repository 使用的 class 和 datasource 里注册的 class 不是同一个实例，最终表现为找不到 metadata。

建议优先保持 API / worker 使用同一个编译后的应用根目录作为 baseDir（例如 dist），然后只通过 detector.ignore / detector.loadDir 控制是否扫描 processor。这样 TypeORM 的目录扫描可以继续写相对公共根目录的路径，例如 entities: ['entity', '**/*.entity.{j,t}s']。

如果一定要把入口拆到 dist/api、dist/worker 这种局部目录，就不要再依赖 entity 的目录扫描，改成你现在这种显式 entities: [ActivityEntity] 会更稳；或者确保 entity 文件也在当前入口 baseDir 可扫描范围内，并且项目里不要同时混用 src 和 dist 两套 entity 路径，避免同一个 entity class 被加载成两份。

[czy88840616]

这个判断是对的。当前实现里 @Processor 的 queueOptions 是在扫描到 processor 后，由 bullmq framework 启动阶段 ensureQueue() 时写入 queueMap 的；如果 producer-only 入口用 detector ignore 掉 processor，getQueue() 为空是预期行为，这时只能显式 createQueue()。

短期建议不要从 @Processor 反向依赖 queue 配置，可以把队列公共配置前移一层：

1. 连接、prefix、通用默认 job 配置放到 bullmq.defaultConnection / bullmq.defaultQueueOptions。
2. 单个业务 queue 的特殊 queueOptions 抽成一个公共配置/常量，API 入口 createQueue("xxx", queueOptions)，worker 入口 @Processor("xxx", jobOptions, workerOptions, queueOptions) 引用同一份配置。

这样虽然 API 入口仍需要 createQueue()，但不会出现两边参数手写不一致。

producer-only 开关目前还没有明确合入时间。后续如果加，我倾向于设计成“允许加载 processor 元数据并创建 queue，但跳过 Worker 创建和 repeat 注册”，这样 producer 进程可以继续通过 getQueue() 拿到带 @Processor queueOptions 的队列，比直接 ignore processor 文件更贴近这个场景。