BullMQ 工作进程卡死问题分析与解决方案

问题现象

在使用 BullMQ 5.41.2 版本构建的 Node.js 队列系统中，工作进程(worker)偶尔会出现停止处理队列任务的情况。具体表现为：

1. 新任务持续被添加到队列中
2. 等待任务数量不断增加
3. 活动任务数量保持不变
4. 只有重启工作进程才能恢复正常

问题诊断

通过日志分析发现，当问题发生时，工作进程虽然显示有活动任务，但实际上这些任务的处理器函数并未真正执行。这表明工作进程可能处于某种阻塞状态。

根本原因

经过深入排查，发现问题根源在于：

1. 并发控制限制：工作进程配置的并发度为3，当3个任务处于活动状态时，系统不会从等待队列中提取新任务
2. 未完成的Promise：某些任务处理器中存在未解决的Promise，导致任务永远处于"活动"状态
3. Redis操作阻塞：部分任务在执行Redis操作时出现阻塞，导致整个工作进程停滞

解决方案

1. 确保Promise正确解决

所有异步操作必须确保Promise能够最终解决(无论成功或失败)。建议采用以下模式：

async function processJob(job) {
  try {
    // 业务逻辑
    await someAsyncOperation();
    return result; // 显式返回结果
  } catch (error) {
    // 捕获并处理错误
    throw error; // 或返回特定错误结果
  }
}

2. 实现任务超时机制

对于可能长时间运行的任务，建议实现超时控制：

const { timeout } = require('bullmq');

async function processWithTimeout(job) {
  try {
    return await timeout(
      async () => {
        // 业务逻辑
      },
      30 * 1000 // 30秒超时
    );
  } catch (error) {
    if (error instanceof timeout.TimeoutError) {
      // 处理超时情况
    }
    throw error;
  }
}

3. 日志追踪优化

对于并发任务日志混乱的问题，建议：

1. 使用追踪ID：为每个任务生成唯一traceId，贯穿所有日志
2. 结构化日志：采用JSON格式输出日志，便于过滤和分析
3. 上下文隔离：利用AsyncLocalStorage等机制保持请求上下文

const { AsyncLocalStorage } = require('async_hooks');
const als = new AsyncLocalStorage();

function withTrace(traceId, fn) {
  return als.run({ traceId }, fn);
}

// 在任务处理器中使用
worker.process(async (job) => {
  return withTrace(job.id, async () => {
    logger.info('开始处理任务');
    // 业务逻辑
  });
});

最佳实践建议

1. 合理设置并发度：根据系统资源和任务特性调整并发数量
2. 完善的错误处理：确保所有可能的错误路径都被捕获和处理
3. 监控与告警：实现队列监控，对长时间运行的任务设置告警
4. 资源清理：确保数据库连接、文件句柄等资源在使用后正确释放
5. 定期健康检查：实现工作进程自检机制，必要时自动重启

总结

BullMQ工作进程卡死问题通常源于未解决的Promise或资源阻塞。通过完善错误处理、实现超时控制、优化日志追踪，可以显著提高系统的稳定性和可维护性。开发者应当特别注意异步操作的完整性，并建立适当的监控机制，确保队列系统能够长期稳定运行。