javascript - 如何确保工作不会在 Bull 中运行两次？

Question

我有两个功能，scheduleScan()和 scan() .
scan()来电scheduleScan() 除了安排新的扫描之外别无其他可做时 , 所以 scheduleScan()可以安排scan() .但是有一个问题，有些作业会运行两次。

我想确保在任何给定时间只处理一项工作。我怎样才能做到这一点？我相信这与 done() 有关, (它在 scan() 中，现在已删除)但我无法提出解决方案。

公牛版:3.12.1

重要后期编辑: scan()调用另一个函数，它们可能会或可能不会调用其他函数，但它们都是同步函数，所以它们只有在自己的工作完成时才调用一个函数，只有一个方法。在“树”的末尾，我称之为，最后一个函数调用 scheduleScan()，但不能同时运行两个作业。每一项工作都从 scan() 开始，顺便说一下，它们以 scheduleScan(stock, period, milliseconds, 'called by file.js') 结尾

export function update(job) {
  // does some calculations, then it may call scheduleScan() or
  // it may call another function, and that could be the one calling
  // scheduleScan() function.
  // For instance, a function like finalize()
}

export function scan(job) {
  update(job)
}


import moment from 'moment'
import stringHash from 'string-hash'
const opts = { redis: { port: 6379, host: '127.0.0.1', password: mypassword' } }
let queue = new Queue('scan', opts)

queue.process(1, (job) => {
  job.progress(100).then(() => {
    scan(job)
  })
})

export function scheduleScan (stock, period, milliseconds, triggeredBy) {
  let uniqueId = stringHash(stock + ':' + period)

  queue.getJob(uniqueId).then(job => {
    if (!job) {
      if (milliseconds) {
        queue.add({ stock, period, triggeredBy }, { delay: milliseconds, jobId: uniqueId }).then(() => {
          // console.log('Added with ms: ' + stock + ' ' + period)
        }).catch(err => {
          if (err) {
            console.log('Can not add because it exists ' + new Date())
          }
        })
      } else {
        queue.add({ stock, period, triggeredBy }, { jobId: uniqueId }).then(() => {
          // console.log('Added without ms: ' + stock + ' ' + period)
        }).catch(err => {
          if (err) {
            console.log('Can not add because it exists ' + new Date())
          }
        })
      }
    } else {
      job.getState().then(state => {
        if (state === 'completed') {
          job.remove().then(() => {
            if (milliseconds) {
              queue.add({ stock, period, triggeredBy }, { delay: milliseconds, jobId: uniqueId }).then(() => {
                // console.log('Added with ms: ' + stock + ' ' + period)
              }).catch(err => {
                if (err) {
                  console.log('Can not add because it exists ' + new Date())
                }
              })
            } else {
              queue.add({ stock, period, triggeredBy }, { jobId: uniqueId }).then(() => {
                // console.log('Added without ms: ' + stock + ' ' + period)
              }).catch(err => {
                if (err) {
                  console.log('Can not add because it exists ' + new Date())
                }
              })
            }
          }).catch(err => {
            if (err) {
              // console.log(err)
            }
          })
        }
      }).catch(err => {
        // console.log(err)
      })
    }
  })
}

Answer 1

问题，我相信是你的scan函数是异步的。所以你的job.progress函数调用 scan然后立即调用done允许队列处理另一个作业。

一个解决方案可能是通过 done回调作为您的 scan 的参数和 scheduleScan函数，并在您完成工作(或出错)后调用它。

另一个(更好的)解决方案可能是确保您始终返回 Promise来自 scan和 scheduleScan ，然后等待 promise 解决，然后调用 done .如果这样做，请确保将所有 promise 返回链接到您的 scheduleScan 中。功能。

queue.process(1, (job, done) => {
  job.progress(100).then(() => {
    scan(job)
        .then(done)
        .catch(done)
  })
})

export function scan() {
   // business logic
   return scheduleScan()
}

// Chain all of your promise returns. Otherwise
// the scan function will return sooner and allow done to be called
// prior to the scheduleScan function finishing it's execution
export function scheduleScan() {
    return queue.getJob(..).then(() => {
        ....
        return queue.add()...
        ....
        return queue.add(...)
            .catch(e => {
                 console.log(e);
                 // propogate errors!
                 throw e;
             })

}