Bull-wachtrij:wanneer een taak mislukt, hoe kunt u voorkomen dat de wachtrij de resterende taken verwerkt?

In bull Het is niet mogelijk om dezelfde taak onmiddellijk na het mislukken te herhalen voordat de volgende taak in de wachtrij wordt opgepakt.

Oplossing:

1. Maak een nieuwe taak en stel de prioriteit in op een waarde die lager is dan het huidige taaktype.
2. Maak de mislukte taak vrij (resolve() of done() )
3. Deze nieuwe baan wordt onmiddellijk opgepikt door de bull voor verwerking.

Voorbeeldcode:In de onderstaande code Job-3 zal mislukken en een nieuwe baan creëren enzovoort totdat "het doel van de baan" op een bepaald moment slaagt.

var Queue = require('bull');

let redisOptions = {
  redis: { port: 6379, host: '127.0.0.1' }
}
var myQueue = new Queue('Linear-Queue', redisOptions);

myQueue.process('Type-1', function (job, done) {
  console.log(`Processing Job-${job.id} Attempt: ${job.attemptsMade}`);
  downloadFile(job, async function (error) {
    if (error) {
      await repeatSameJob(job, done);
    } else {
      done();
    }
  });
});

async function repeatSameJob(job, done) {
  let newJob = await myQueue.add('Type-1', job.data, { ...{ priority: 1 }, ...job.opts });
  console.log(`Job-${job.id} failed. Creating new Job-${newJob.id} with highest priority for same data.`);
  done(true);
}

function downloadFile(job, done) {
  setTimeout(async () => {
    done(job.data.error)
  }, job.data.time);
}

myQueue.on('completed', function (job, result) {
  console.log("Completed: Job-" + job.id);
});

myQueue.on('failed', async function (job, error) {
  console.log("Failed: Job-" + job.id);
});

let options = {
  removeOnComplete: true, // removes job from queue on success
  removeOnFail: true // removes job from queue on failure
}

for (let i = 1; i <= 5; i++) {
  let error = false;
  if (i == 3) { error = true; }

  setTimeout(i => {
    let jobData = {
      time: i * 2000,
      error: error,
      description: `Job-${i}`
    }
    myQueue.add('Type-1', jobData, options);
  }, i * 2000, i);
}

Uitgang: