feat: implement BullMQ task queue integration with configuration options for concurrency, retries, and Redis setup in backend-nodejs

Author: erweixin

backend-nodejs/cmd/server/main.ts

@@ -24,6 +24,7 @@ import {
   registerDomainRoutes,
 } from '../../infrastructure/bootstrap/server.js';
 import { initDependencies } from '../../infrastructure/bootstrap/dependencies.js';
+import { startQueueWorkers } from '../../infrastructure/queue/worker_bootstrap.js';
 import type { RedisClientType } from 'redis';
 
 async function main() {
@@ -100,10 +101,17 @@ async function main() {
 
   // 7. 初始化依赖注入容器
   console.log('🏗️  Initializing dependency injection container...');
-  const container = initDependencies(config, db, redis);
+  const container = await initDependencies(config, db, redis);
   console.log('✅ Dependencies initialized');
 
-  // 8. 注册领域路由
+  // 8. 启动队列 Worker（如果启用）
+  if (container.workerManager && config.queue.enabled && container.queueProcessors) {
+    console.log('⚙️  Starting queue workers...');
+    startQueueWorkers(container.workerManager, container.queueProcessors, config);
+    console.log('✅ Queue workers started');
+  }
+
+  // 9. 注册领域路由
   console.log('📚 Registering domain routes...');
   await registerDomainRoutes(
     fastify,
@@ -116,7 +124,7 @@ async function main() {
     redis
   );
 
-  // 9. 启动服务器
+  // 10. 启动服务器
   const address = `http://${config.server.host}:${config.server.port}`;
   try {
     await fastify.listen({
@@ -134,7 +142,7 @@ async function main() {
     process.exit(1);
   }
 
-  // 10. 优雅关闭
+  // 11. 优雅关闭
   const shutdown = async (signal: string) => {
     console.log(`\n🛑 Received ${signal}, shutting down gracefully...`);
     try {
@@ -143,6 +151,13 @@ async function main() {
       if (redis) {
         await closeRedisConnection(redis);
       }
+      // 关闭队列
+      if (container.queueClient) {
+        await container.queueClient.close();
+      }
+      if (container.workerManager) {
+        await container.workerManager.stopAll();
+      }
       // 关闭事件总线
       await container.eventBus.close();
       // 刷新日志缓冲区

backend-nodejs/env.example

@@ -56,7 +56,19 @@ TRACING_ENDPOINT=http://localhost:4318
 # ANTHROPIC_API_KEY=your-anthropic-api-key
 
 # ==================== 队列配置（可选，用于 BullMQ）====================
-# QUEUE_REDIS_HOST=localhost
-# QUEUE_REDIS_PORT=6379
-# QUEUE_REDIS_PASSWORD=
+# 队列功能开关
+QUEUE_ENABLED=true
+
+# 队列使用的 Redis DB（与缓存分离，避免冲突）
+# 缓存使用 REDIS_DB=0，队列使用 QUEUE_REDIS_DB=1
+QUEUE_REDIS_DB=1
+
+# Worker 配置
+QUEUE_WORKER_CONCURRENCY=10        # Worker 并发数（同时处理的任务数）
+QUEUE_WORKER_MAX_RETRIES=3        # 最大重试次数
+QUEUE_WORKER_RETRY_DELAY=5000     # 重试延迟（毫秒）
+
+# 注意：队列使用与 Redis 缓存相同的 Redis 实例
+# 但使用不同的 DB 编号（REDIS_DB vs QUEUE_REDIS_DB）
+# 队列配置会从上面的 REDIS_HOST、REDIS_PORT、REDIS_PASSWORD 读取连接信息
 

backend-nodejs/infrastructure/bootstrap/dependencies.ts

@@ -14,6 +14,9 @@ import type { Kysely } from 'kysely';
 import type { RedisClientType } from 'redis';
 import type { Database } from '../persistence/postgres/database.js';
 import type { Config } from '../config/config.js';
+import type { QueueClient, WorkerManager } from '../queue/index.js';
+import type { QueueProcessorsResult } from '../queue/bootstrap.js';
+import { bootstrapQueueProcessors } from '../queue/bootstrap.js';
 
 // Repository 实现
 import { TaskRepositoryImpl } from '../../domains/task/repository/task_repo.js';
@@ -37,13 +40,20 @@ import { createAuthMiddleware } from '../middleware/auth.js';
 // Event Bus
 import { createEventBus, type EventBus } from '../../domains/shared/events/event_bus.js';
 
+// Queue
+import { createQueueClient } from '../queue/client.js';
+import { createWorkerManager } from '../queue/worker.js';
+
 /**
  * 应用依赖容器
  * 包含所有领域的 Handler Dependencies 和 Middleware
  */
 export interface AppContainer {
   // 基础设施
   eventBus: EventBus;
+  queueClient: QueueClient | null;
+  workerManager: WorkerManager | null;
+  queueProcessors: QueueProcessorsResult | null; // 队列处理器引导结果
 
   // Auth 领域
   authHandlerDeps: AuthHandlerDependencies;
@@ -64,18 +74,40 @@ export interface AppContainer {
  * @param _redis Redis 连接（可选，保留用于未来扩展）
  * @returns 应用依赖容器
  */
-export function initDependencies(
+export async function initDependencies(
   config: Config,
   db: Kysely<Database>,
   _redis: RedisClientType | null = null
-): AppContainer {
+): Promise<AppContainer> {
   // ============================================
   // Infrastructure Layer（基础设施层）
   // ============================================
 
   // 事件总线（用于领域间通信）
   const eventBus = createEventBus();
 
+  // 队列客户端和 Worker 管理器（如果启用）
+  let queueClient: QueueClient | null = null;
+  let workerManager: WorkerManager | null = null;
+  let queueProcessors: QueueProcessorsResult | null = null;
+
+  if (config.queue.enabled && _redis) {
+    const queueRedisConfig = {
+      host: config.redis.host,
+      port: config.redis.port,
+      password: config.redis.password || undefined,
+      db: config.queue.redisDb,
+    };
+
+    queueClient = createQueueClient(queueRedisConfig);
+    workerManager = createWorkerManager(queueRedisConfig);
+
+    // 引导并注册所有队列处理器（领域自注册模式）
+    queueProcessors = await bootstrapQueueProcessors({
+      includeExamples: config.server.env === 'development',
+    });
+  }
+
   // ============================================
   // Repository Layer（基础设施层）：数据访问
   // ============================================
@@ -139,6 +171,9 @@ export function initDependencies(
 
   return {
     eventBus,
+    queueClient,
+    workerManager,
+    queueProcessors,
     authHandlerDeps,
     authMiddleware,
     userHandlerDeps,

backend-nodejs/infrastructure/config/config.ts

@@ -74,6 +74,15 @@ const ConfigSchema = z.object({
     password: z.string().default(''),
     db: z.coerce.number().int().nonnegative().default(0),
   }),
+  queue: z.object({
+    enabled: z.coerce.boolean().default(true),
+    redisDb: z.coerce.number().int().nonnegative().default(1), // 队列使用 DB 1，与缓存分离
+    worker: z.object({
+      concurrency: z.coerce.number().int().positive().default(10),
+      maxRetries: z.coerce.number().int().nonnegative().default(3),
+      retryDelay: z.coerce.number().int().nonnegative().default(5000), // 毫秒
+    }),
+  }),
   jwt: z.object({
     secret: z
       .string()
@@ -157,6 +166,15 @@ export function loadConfig(): Config {
         password: process.env.REDIS_PASSWORD,
         db: process.env.REDIS_DB,
       },
+      queue: {
+        enabled: process.env.QUEUE_ENABLED,
+        redisDb: process.env.QUEUE_REDIS_DB,
+        worker: {
+          concurrency: process.env.QUEUE_WORKER_CONCURRENCY,
+          maxRetries: process.env.QUEUE_WORKER_MAX_RETRIES,
+          retryDelay: process.env.QUEUE_WORKER_RETRY_DELAY,
+        },
+      },
       jwt: {
         secret: process.env.JWT_SECRET,
         accessTokenExpiry: process.env.JWT_ACCESS_TOKEN_EXPIRY,