Flows
Flows are a brand new feature in BullMQ, and although is implemented on a stable foundation there could be some unknown issues.
BullMQ supports parent - child relationships between jobs. The basic idea is that a parent job will not be moved to the wait status, i.e. can be picked up by a worker, until all its children jobs have been processed successfully. Apart from that, a parent or a child job are no different from regular jobs.
This functionality enables the creation of flows where jobs are the node of trees of arbitrary depth.
Flows are added to a queue using the "FlowProducer" class.
In order to create "flows" you must use the FlowProducer class. The method add accepts an object with the following interface:
interface FlowJob {
name: string;
queueName: string;
data?: any;
prefix?: string;
opts?: Omit<JobsOptions, 'parent' | 'repeat'>;
children?: FlowJob[];
}
So we can add a flow like this one:
TypeScript
Python
import { FlowProducer } from 'bullmq';
// A FlowProducer constructor takes an optional "connection"
// object otherwise it connects to a local redis instance.
const flowProducer = new FlowProducer();
const flow = await flowProducer.add({
name: 'renovate-interior',
queueName: 'renovate',
children: [
{ name: 'paint', data: { place: 'ceiling' }, queueName: 'steps' },
{ name: 'paint', data: { place: 'walls' }, queueName: 'steps' },
{ name: 'fix', data: { place: 'floor' }, queueName: 'steps' },
],
});
from bullmq import FlowProducer
# A FlowProducer constructor takes an optional "connection"
# object otherwise it connects to a local redis instance.
flowProducer = FlowProducer()
flow = await flowProducer.add({
"name": "renovate-interior",
"queueName": "renovate",
"children": [
{ "name": "paint", "data": { "place": "ceiling" }, "queueName": "steps" },
{ "name": "paint", "data": { "place": "walls" }, "queueName": "steps" },
{ "name": "fix", "data": { "place": "floor" }, "queueName": "steps" },
],
})
The above code will add atomically 4 jobs, one to the "renovate" queue and 3 to the "steps" queue. When the 3 jobs in the "steps" queue are completed, the parent job in the "renovate" queue will be processed as a regular job.
The above call will return instances for all the jobs added to the queue.
Note that the parent queue does not need to be the same queue as the one used for the children.
When the parent job is processed it is possible to access the results generated by its child jobs. For example, lets assume the following worker for the child jobs:
TypeScript
Python
import { Worker } from 'bullmq';
const stepsWorker = new Worker('steps', async job => {
await performStep(job.data);
if (job.name === 'paint') {
return 2500;
} else if (job.name === 'fix') {
return 1750;
}
});
from bullmq import Worker
async def process(job: Job, token: str):
await performStep(job.data)
if job.name == 'paint':
return 2500
elif job.name == 'fix':
return 1750
stepsWorker = Worker("steps", process, {"connection": connection})
We can implement a parent worker that sums the costs of the children's jobs using the "getChildrenValues" method. This method returns an object with job keys as keys and the result of that given job as a value:
import { Worker } from 'bullmq';
const renovateWorker = new Worker('renovate', async job => {
const childrenValues = await job.getChildrenValues();
const totalCosts = Object.values(childrenValues).reduce(
(prev, cur) => prev + cur,
0,
);
await sendInvoice(totalCosts);
});
It is possible to add as deep job hierarchies as needed, see the following example where jobs are depending on each other, this allows serial execution of jobs:
TypeScript
Python
import { FlowProducer } from 'bullmq';
const flowProducer = new FlowProducer();
const queueName = 'assembly-line';
const chain = await flowProducer.add({
name: 'car',
data: { step: 'engine' },
queueName,
children: [
{
name: 'car',
data: { step: 'wheels' },
queueName,
children: [{ name: 'car', data: { step: 'chassis' }, queueName }],
},
],
});
from bullmq import FlowProducer
flowProducer = FlowProducer()
queueName = 'assembly-line'
chain = await flowProducer.add({
"name": "car",
"data": { "step": "engine" },
"queueName": queueName,
"children": [
{
"name": "car",
"data": { "step": "wheels" },
"queueName": queueName,
"children": [{ "name": "car", "data": { "step": "chassis" }, "queueName": queueName }],
},
],
})
In this case one job will be processed after the previous one has been completed.
The order of processing would be: 'chassis', 'wheels' and finally 'engine'.
There are some special getters that can be used in order to get jobs related to a flow. First, we have a method in the Job class to get all the dependencies for a given job:
const dependencies = await job.getDependencies();
it will return all the direct dependencies, i.e. the children of a given job.
The Job class also provides another method that we presented above to get all the values produced by the children of a given job:
const values = await job.getChildrenValues();
Also, a new property is available in the Job class, parentKey, with a fully qualified key for the job parent.
Finally, there is also a new state where a job can be in, "waiting-children", for parent jobs that have not yet had their children completed:
const state = await job.getState();
// state will be "waiting-children"
When adding a flow it is also possible to provide an extra options object "queueOptions", where you can add your specific options for every queue that is used in the flow. These options would affect each one of the jobs that are added to the flow using the FlowProducer.
import { FlowProducer } from 'bullmq';
const flowProducer = new FlowProducer();
const queueName = 'assembly-line';
const chain = await flowProducer.add(
{
name: 'car',
data: { step: 'engine' },
queueName,
children: [
{
name: 'car',
data: { step: 'wheels' },
queueName,
},
],
},
{
queuesOptions: {
[queueName]: {
defaultJobOptions: {
removeOnComplete: true,
},
},
},
},
);
Queue options are defined in the context of their instances. You should provide your configurations in the second parameter in order to avoid unexpected behaviors.
BullMQ also provides seamless removal functionality for jobs that are part of a flow.
When removing a job that is part of the flow there are several important considerations:
- 1.If a parent job is removed, all its children will also be removed.
- 2.If a child job is removed, its parent dependency to said child is also removed, and if the child was the last child in the dependencies list, the parent job will be completed.
- 3.Since a job can be both a parent and a child in a large flow, both 1 and 2 will occur if removing such a job.
- 4.If any of the jobs that would be removed happen to be locked, none of the jobs will be removed, and an exception will be thrown.
Apart from the considerations above, removing a job can simply be done by either using the Job or the Queue class:
await job.remove();
// or
await queue.remove(job.id);
Last modified 27d ago