node --version 

mkdir bullmq-demo cd bullmq-demo npm init -y npm install bullmq ioredis npm install -D typescript tsx @types/node 

{ "compilerOptions": { "target": "ES2022", "module": "nodenext", "moduleResolution": "nodenext", "types": ["node"], "strict": true, "esModuleInterop": true, "skipLibCheck": true, "noEmit": true }, "include": ["src"] } 

npx tsc --noEmit 

# start Redis in the background on port 6379 docker run -d --name bullmq-redis -p 6379:6379 redis:7-alpine # check it is up (should print PONG) docker exec bullmq-redis redis-cli ping # when you are done: stop and remove the container docker stop bullmq-redis docker rm bullmq-redis 

services: redis: image: redis:7-alpine # Turn on the append-only file, so scheduled jobs survive a restart # (the image's default is periodic snapshots only; see Part 4) command: redis-server --appendonly yes ports: - '6379:6379' volumes: - redis-data:/data volumes: redis-data: 

# start Redis in the background docker compose up -d # check it is up (should print PONG) docker compose exec redis redis-cli ping # when you are done: stop it (add -v to also delete the stored data) docker compose down 

bullmq-demo/ docker-compose.yaml (only if you use Option B) package.json tsconfig.json src/ connection.ts worker.ts scheduler.ts workerManualRetry.ts 

// src/connection.ts // Shared Redis connection for the article demo export const connection = { host: '127.0.0.1', port: 6379, }; export const QUEUE_NAME = 'demo-reminders'; // Shape of the job body (what the scheduler passes as `data`). // The retry settings travel WITH the job, so the manual retry worker // (workerManualRetry.ts) can decide on its own whether to reschedule and how // long to wait. export interface ReminderJobData { orderId: string; reason: string; attempt: number; // which attempt this is, 1-based maxAttempts: number; // how many attempts to make in total before giving up retryDelayMs: number; // how long to wait before the next attempt } 

// src/worker.ts import { Worker } from 'bullmq'; import { connection, QUEUE_NAME } from './connection'; // The WORKER is the consumer side of the queue: it pulls jobs from Redis and processes them. // Basically the worker contains the business logic to process the job's data, like sending emails, calling APIs and the database // It does not reschedule or mark the job itself: returning normally marks the job completed, and throwing marks it failed — BullMQ then retries it according to the job's attempts/backoff options // Keep it running in one terminal, then schedule jobs from another. // The worker continuously pulls jobs for processing // // A Worker is bound to exactly ONE queue. If you have several queues you need to create a separate worker instance for every queue name // The first argument is the queue name (Example: "DEMO_QUEUE") // The second argument is the async function which contains the business logic // The third argument is the connection configuration const worker = new Worker( QUEUE_NAME, async (job) => { // job.id — unique id of this job in the queue // job.name — the job's name (set by queue.add('reminder', ...) in // scheduler.ts); use it to route different job types handled // by the same worker // job.data — the payload passed when the job was scheduled console.log('Processing queue:', QUEUE_NAME); console.log(`[${new Date().toLocaleTimeString()}] processing job ${job.id}:`, job.data); }, { connection }, ); // Lifecycle events. The worker keeps a long-lived connection to Redis and emits // an event for every job it finishes. worker.on('completed', (job) => console.log(` done: ${job.id}`)); // job succeeded worker.on('failed', (job, err) => console.error(` failed: ${job?.id} — ${err.message}`)); // job threw // Redis connection problems are reported here. Without this listener, BullMQ // prints a raw stack trace on every reconnect attempt. worker.on('error', (err) => console.error(' worker error:', err.message)); 

// src/scheduler.ts import { Queue } from 'bullmq'; import { connection, QUEUE_NAME } from './connection'; // A Queue is the producer-side handle to a queue in Redis: you use it to ADD // jobs and to look them up / remove them. It does not process anything — that's // the worker's job. // arg 1 — queue name (must match the worker's queue name) // arg 2 — options, including the Redis connection const queue = new Queue(QUEUE_NAME, { connection }); // Redis connection problems are reported here. Without this listener, BullMQ // prints a raw stack trace on every reconnect attempt. queue.on('error', (err) => console.error('queue error:', err.message)); // The SCHEDULER enqueues a delayed job, or cancels one before it runs. // npx tsx src/scheduler.ts -> schedule a job (fires in 5s) // npx tsx src/scheduler.ts cancel <id> -> cancel that job async function main() { // Read CLI args: e.g. `... scheduler.ts cancel 42` -> action="cancel", jobId="42" const [action, jobId] = process.argv.slice(2); if (action === 'cancel') { // getJob(id) looks a job up by id. Returns the Job, or undefined if it has // already run or was removed. const job = await queue.getJob(jobId); if (job) { // remove() deletes the job from the queue. For a delayed job that hasn't // run yet, this cancels it — the worker will never see it. await job.remove(); console.log(`cancelled job ${jobId} — it will not run`); } else { console.log(`no job ${jobId} found (already ran or removed)`); } } else { const delayMs = 5000; // add() puts a new job on the queue. // arg 1 — job name ('reminder'): a label the worker can switch on to // handle different job types on the same queue. // arg 2 — data: the payload, must be JSON-serializable. The worker reads // this as job.data. The `attempt` / `maxAttempts` / `retryDelayMs` // fields are only used by the manual retry worker // (workerManualRetry.ts); the basic worker just logs them. // arg 3 — options: here `delay` (ms) holds the job back so it only becomes // available to the worker after that time. Other options live here // too: `attempts`, `backoff`, `jobId`, `priority`, etc. (Recurring // jobs are not an add() option — use queue.upsertJobScheduler().) const job = await queue.add( 'reminder', { orderId: 'order-123', reason: 'unpaid', attempt: 1, maxAttempts: 3, retryDelayMs: 3000, }, { delay: delayMs }, ); console.log(`scheduled job ${job.id} to run in ${delayMs / 1000}s`); console.log(`cancel it with: npx tsx src/scheduler.ts cancel ${job.id}`); } // Close the connection so this short-lived script can exit (the worker, by // contrast, stays open to keep processing). await queue.close(); } main(); 

// src/workerManualRetry.ts // // MANUAL RETRY EXAMPLE — an alternative to worker.ts. // // Instead of letting BullMQ retry (via `attempts` + `backoff` in the job // options), the retry settings travel in the JOB BODY and this worker decides // for itself whether to try again. On failure it adds a NEW job to the same // queue with attempt + 1, delayed by retryDelayMs. // // Run EITHER this worker OR worker.ts — not both. They consume the same queue, // so if both run they will compete for the same jobs. // // Trade-offs vs BullMQ's built-in retry: // * the retry is a NEW job with a NEW id (built-in retry reuses the same job // id and increments job.attemptsMade) // * because we CATCH the error, BullMQ never sees a failure — every attempt // is recorded as 'completed' and the 'failed' event never fires // Use this only when you need that control; otherwise prefer `attempts` + `backoff`. import { Queue, Worker } from 'bullmq'; import { connection, QUEUE_NAME, ReminderJobData } from './connection'; // A Queue handle, so the worker can put the NEXT attempt back on the same queue. // Only needed because we are retrying manually. const queue = new Queue(QUEUE_NAME, { connection }); queue.on('error', (err) => console.error(' queue error:', err.message)); // Simulated work — this is the part that can fail in real life (SMTP down, // payment API timing out, etc.). It fails on attempt 1 and succeeds from // attempt 2, so the retry path is easy to watch in the demo. async function sendReminder(data: ReminderJobData): Promise<void> { if (data.attempt === 1) { throw new Error('payment provider timed out'); } console.log(` reminder sent for ${data.orderId} (${data.reason})`); } const worker = new Worker<ReminderJobData>( QUEUE_NAME, async (job) => { // The retry settings come from the job body, not the job options. const { attempt, maxAttempts, retryDelayMs } = job.data; console.log('Processing queue:', QUEUE_NAME); console.log( `[${new Date().toLocaleTimeString()}] job ${job.id} — attempt ${attempt}/${maxAttempts}`, ); try { await sendReminder(job.data); console.log(` success on attempt ${attempt}`); } catch (err) { const reason = (err as Error).message; if (attempt < maxAttempts) { // RESCHEDULE: add a NEW job to the same queue carrying an incremented // attempt counter, delayed by retryDelayMs. That delayed job becomes // "attempt 2", and this one finishes normally. // arg 1 — same job name, so the same worker logic handles it // arg 2 — the same body, with attempt bumped by 1 // arg 3 — delay: wait this long before the retry becomes available await queue.add(job.name, { ...job.data, attempt: attempt + 1 }, { delay: retryDelayMs }); console.log( ` attempt ${attempt} failed (${reason}) — retrying in ${retryDelayMs / 1000}s as attempt ${attempt + 1}`, ); } else { // No attempts left: stop here instead of scheduling another job. console.error(` attempt ${attempt} failed (${reason}) — no attempts left, giving up`); } } }, { connection }, ); // NOTE: 'completed' fires for every attempt here — including the ones whose // work failed — because the error was caught above. 'failed' only fires for // errors we did not catch. worker.on('completed', (job) => console.log(` done: ${job.id}`)); worker.on('failed', (job, err) => console.error(` failed: ${job?.id} — ${err.message}`)); // Redis connection problems are reported here. Without these listeners, BullMQ // prints a raw stack trace on every reconnect attempt. worker.on('error', (err) => console.error(' worker error:', err.message)); 

const job = await queue.add( 'reminder', { orderId: 'order-123', reason: 'unpaid', attempt: 1, maxAttempts: 3, retryDelayMs: 3000, }, { delay: delayMs, attempts: 3, backoff: { type: 'exponential', delay: 1000 } }, ); 

// src/workerBuiltInRetry.ts // // BUILT-IN RETRY EXAMPLE — the recommended way. // // The worker simply THROWS when the work fails. BullMQ sees the failure and // retries the SAME job automatically, using the `attempts` and `backoff` // options given when the job was added (see scheduler.ts). // // Run EITHER this worker OR worker.ts / workerManualRetry.ts — not more than // one. They consume the same queue, so they would compete for the same jobs. import { Worker } from 'bullmq'; import { connection, QUEUE_NAME } from './connection'; // Simulated work — fails on the first attempt and succeeds on the second, so // the retry is easy to watch. Source：https://dev.to/tarun_koshti_122/beyond-caching-scheduling-jobs-with-redis-nodejs-and-bullmq-31a5
My favorites