1

我一直在为我们的项目广泛使用 Bull 库。几天前，我们在处理 Bull Queue 的作业时发现了这个问题。我们能够在 Bull Queue 中添加作业，但它无法处理该作业。当我签入工作组时，我们注意到作业延迟/失败，错误响应为空。我认为这是因为工作进程没有附加到队列中。如果我在一段时间后尝试重新启动进程，则相同的代码可以正常工作。我不确定这是 Redis 或 Bull 的问题还是我使用它的方式。问题是我可以添加作业，但作业未处理。（仅在某些情况下出乎意料）

示例代码片段：

const BullQueue = require('bull');
class Queue {
    constructor(name, connectionName = DBConstant.localRedisConnectionName) {
        const options = {};
        options.redis = DBConnectionUseCase.getRedisConnectionOptions(connectionName);
        this._queue = new BullQueue(name, options);
        this._connectionName = connectionName;
        this.queueName = name;
   }
    async initProcessor() {
        try {
            //TODO: We have noticed that in some cases process does not get attached to Queue.
            //We tried adding await here and check result but did not get anything. This needs to be debug.
            this._queue.process((job, done) => {
                this.process(job, done);
            });
        }catch(error) {
            console.log(`Worker :: ${this.queueName} :: Exception in processor initialisation :: connectionName ::  ${this._connectionName} :: Error :: ${error.message} :: ${JSON.stringify(error)}`);
        }
        console.log(`Worker :: ${this.queueName} :: processor initialised :: connectionName :: ${this._connectionName}`);
    }

    getQueueName() {
        return this.queueName;
    }

    addJob(data, options) {
        console.log(`Worker :: ${this.queueName} :: Job added in Queue :: ${this._connectionName}`);
        return this._queue.add(data, options);
    }
}

class FetchCustomerData extends Queue{
  constructor() {
    super(QUEUE_NAME);
  }

  /**
   * 
   */
  static getInstance() {
    if(!queueInstance) {
      queueInstance = new FetchCustomerData();
    }
    return queueInstance;
  }

  /**
   * 
   */
  initDefaultJob() {
    const data = {};
    const options = {
        // This cron will run in everyday at 12:30AM ISE
        repeat: {
          cron: '0 20 * * *'
        }
    };

    this.addJob(data, Object.assign({},options, constant.BULL_JOB_OPTIONS));
    console.log(`Worker :: ${QUEUE_NAME} :: initial job added :: options :: ${JSON.stringify(Object.assign({},options, constant.BULL_JOB_OPTIONS))}`);
  }

  /**
   * 
  */
  async process(job, done) {
    try {
        ...
        return done(nu;;);
    }catch(error) {
      console.error(`FetchActivationalUCC.process :: ${uuid} :: Exception :: ${JSON.stringify(error)}`);
      return done(error);
    }
  }
}

我如何使用它：

FetchCustomerData.initProcessor();
const res = await FetchCustomerData.addJob({"key": 123});

包装清单：

"bull": "3.7.0"
"ioredis": "4.9.0"
"node": v10.15.0
4

0 回答 0