·QueueHub Team·18 min read

Advanced BullMQ Worker Patterns — Multi-Queue Processing, Dynamic Concurrency & Custom Metrics

BullMQWorker PatternsMulti-Queue WorkersDynamic ConcurrencyPrometheusOpenTelemetryKubernetesProduction

If you already know how to spin up a BullMQ Worker, configure concurrency, and attach event listeners, you're past the tutorial phase. But production introduces questions the "getting started" guide doesn't answer: How do I process jobs from multiple queues in a single worker without spawning dozens of processes? How do I change concurrency at 3 AM without a deploy? How do I expose worker-level metrics that my Prometheus stack actually understands? And how do I manage the worker lifecycle — pausing, draining, health-checking — when Kubernetes is in control?

This post answers those questions. Every section is written for developers who already know BullMQ's basics and need battle-tested patterns for the messy reality of production. You'll learn multi-queue routing, CPU-aware adaptive concurrency, custom Prometheus instrumentation, worker health probes, and the full lifecycle API beyond close().

Multi-Queue Workers — Routing Jobs from Different Sources Into One Processor

The simplest deployment is one queue, one worker, one job type. But real systems have many job types — emails, report generation, image processing, webhook delivery. The naive approach is one worker instance per queue. That works, but it burns through Redis connections, duplicates lifecycle code, and makes resource pooling impossible.

A multi-queue worker — a single Worker instance that processes jobs from multiple sources — solves all three problems. There are two common patterns.

Pattern 1: Shared Queue with Dispatch-by-Job-Name

The cleanest approach: push all job types into one queue and use the job name field as a routing key. BullMQ's queue.add(jobName, data) lets producers tag jobs by type, and a single processor function dispatches on that name:

import { Worker, Job } from 'bullmq';
import { Redis } from 'ioredis';

const connection = new Redis({
  host: process.env.REDIS_HOST,
  maxRetriesPerRequest: null,
});

type EmailPayload = { to: string; template: string };
type ReportPayload = { reportId: string; format: 'pdf' | 'csv' };
type ImagePayload = { sourceUrl: string; sizes: number[] };

async function processEmail(job: Job<EmailPayload>) {
  const result = await sendEmail(job.data.to, job.data.template);
  return { sent: true, messageId: result.id };
}

async function processReport(job: Job<ReportPayload>) {
  const url = await generateReport(job.data.reportId, job.data.format);
  return { url };
}

async function processImage(job: Job<ImagePayload>) {
  const urls = await resizeImages(job.data.sourceUrl, job.data.sizes);
  return { urls };
}

const worker = new Worker<EmailPayload | ReportPayload | ImagePayload>(
  'shared-work-queue',
  async (job) => {
    switch (job.name) {
      case 'email':
        return processEmail(job as Job<EmailPayload>);
      case 'report':
        return processReport(job as Job<ReportPayload>);
      case 'image':
        return processImage(job as Job<ImagePayload>);
      default:
        throw new Error(`Unknown job type: ${job.name}`);
    }
  },
  { connection, concurrency: 10 }
);

worker.on('ready', () => console.log('Shared worker ready'));
worker.on('error', (err) => console.error('Worker error:', err));

When to use this pattern: You have related job types that share the same SLA, retention policy, and processing infrastructure. The single queue simplifies monitoring — one depth gauge, one rate metric.

Pattern 2: Coordinator Worker with Separate Queues

Sometimes different job types need different SLAs — email delivery must be fast, but report generation can wait. Keeping them in separate queues gives you independent rate limits, concurrency settings, and retention policies. A coordinator worker bridges them:

import { Worker, Queue, Job } from 'bullmq';

const emailQueue = new Queue('email-high-priority', { connection });
const reportQueue = new Queue('report-low-priority', { connection });
const imageQueue = new Queue('image-batch', { connection });

// A single worker drains a shared "work" queue
// An enqueue helper routes jobs from producer queues to the shared consumer
async function enqueueToShared(
  sourceQueueName: string,
  jobName: string,
  data: unknown,
  options?: { priority?: number; attempts?: number }
): Promise<void> {
  const shared = new Queue('shared-work', { connection });
  await shared.add(jobName, { _source: sourceQueueName, ...(data as Record<string, unknown>) }, {
    priority: options?.priority ?? 0,
    attempts: options?.attempts ?? 3,
  });
}

When to use this pattern: Job types have different QoS requirements, you need per-queue rate limiting, or teams own different producer queues but share a pool of consumers.

Pro tip: Use job.queueName (available on the Job object) inside the processor to detect which original queue a job came from. This lets you branch on both job.name and source queue for fine-grained routing.

Dynamic Concurrency Adjustment at Runtime

Setting concurrency: 10 in the Worker constructor and forgetting about it works — until it doesn't. A traffic spike overwhelms your downstream API. A co-located process hogs CPU. A memory leak makes your worker heap grow uncontrollably. Static concurrency has no escape valve.

Here's the good news: worker.concurrency is a writable property. You can change it while the worker is running without restarting. Active jobs finish at their current concurrency level; new jobs are picked up at the new rate.

CPU-Aware Adaptive Concurrency

A simple feedback loop measures system load and adjusts concurrency accordingly:

import { Worker, Job } from 'bullmq';
import os from 'os';

class AdaptiveWorker {
  private worker: Worker;
  private currentConcurrency: number;
  private minConcurrency = 1;
  private maxConcurrency: number;
  private intervalHandle?: NodeJS.Timeout;

  constructor(
    queueName: string,
    processor: (job: Job) => Promise<unknown>,
    opts: {
      connection: Redis;
      baseConcurrency?: number;
      maxConcurrency?: number;
      checkIntervalMs?: number;
    }
  ) {
    this.currentConcurrency = opts.baseConcurrency ?? 10;
    this.maxConcurrency = opts.maxConcurrency ?? 50;

    this.worker = new Worker(queueName, processor, {
      connection: opts.connection,
      concurrency: this.currentConcurrency,
    });

    this.worker.on('error', (err) =>
      console.error('[AdaptiveWorker] error:', err)
    );

    this.startMonitoring(opts.checkIntervalMs ?? 10_000);
  }

  private startMonitoring(intervalMs: number): void {
    this.intervalHandle = setInterval(() => {
      const target = this.computeTargetConcurrency();
      if (target !== this.currentConcurrency) {
        console.log(
          `[AdaptiveWorker] concurrency ${this.currentConcurrency} -> ${target} ` +
          `(cpu=${os.loadavg()[0].toFixed(2)}, ` +
          `mem=${(process.memoryUsage().heapUsed / 1024 / 1024).toFixed(0)}MB)`
        );
        this.worker.concurrency = target;
        this.currentConcurrency = target;
      }
    }, intervalMs);
  }

  private computeTargetConcurrency(): number {
    const cpuLoad = os.loadavg()[0] / os.cpus().length;
    const heapUsedMB = process.memoryUsage().heapUsed / 1024 / 1024;
    const heapTotalMB = process.memoryUsage().heapTotal / 1024 / 1024;
    const memRatio = heapTotalMB > 0 ? heapUsedMB / heapTotalMB : 0;

    // Multiplicative decrease when overloaded
    if (cpuLoad > 0.8 || memRatio > 0.85) {
      return Math.max(
        this.minConcurrency,
        Math.floor(this.currentConcurrency * 0.5)
      );
    }

    // Additive increase when healthy
    if (cpuLoad < 0.5 && memRatio < 0.6 && this.currentConcurrency < this.maxConcurrency) {
      return Math.min(this.maxConcurrency, this.currentConcurrency + 2);
    }

    return this.currentConcurrency;
  }

  async close(): Promise<void> {
    clearInterval(this.intervalHandle);
    await this.worker.close();
  }
}

API-429-Driven Throttling

When a downstream API returns 429 Too Many Requests, static concurrency is doubly dangerous — not only are you generating rate-limit errors, you're burning retry attempts on jobs that will keep failing until the API recovers. Tune concurrency dynamically based on API responses:

class RateLimitAwareWorker {
  private worker: Worker;
  private baseConcurrency: number;
  private cooldownUntil = 0;

  constructor(queueName: string, connection: Redis, baseConcurrency = 10) {
    this.baseConcurrency = baseConcurrency;

    this.worker = new Worker(
      queueName,
      async (job) => {
        // During cooldown, run at 25% capacity
        if (Date.now() < this.cooldownUntil) {
          this.worker.concurrency = Math.max(1, Math.floor(this.baseConcurrency * 0.25));
        } else {
          this.worker.concurrency = this.baseConcurrency;
        }

        try {
          const response = await callExternalApi(job.data);
          return response.data;
        } catch (err: any) {
          if (err.response?.status === 429) {
            const retryAfter = parseInt(
              err.response.headers['retry-after'] ?? '30', 10
            );
            this.cooldownUntil = Date.now() + retryAfter * 1000;
          }
          throw err; // Let BullMQ handle the retry
        }
      },
      { connection, concurrency: baseConcurrency }
    );
  }
}

Pro tip: Setting worker.concurrency to 0 effectively pauses the worker without calling worker.pause(). Jobs already in flight finish normally, but no new jobs are pulled from Redis. This is useful when you need a non-blocking pause that respects the current unit of work but doesn't involve Redis state changes.

Custom Worker Instrumentation and Production Metrics

BullMQ's built-in Metrics system gives you queue-level throughput. But when you're debugging a production incident, queue-level aggregates hide the real story: which worker is slow, which job type is failing, and how saturated each worker process is.

Building a custom metrics layer — exposed on a /metrics endpoint via Prometheus — gives you worker-level observability.

Prometheus Instrumentation with prom-client

This pattern wraps a BullMQ worker with a Prometheus metrics layer using the prom-client library:

import { Worker, Job } from 'bullmq';
import client from 'prom-client';
import express from 'express';
import os from 'os';

// --- Metric definitions ---
const jobDurationHistogram = new client.Histogram({
  name: 'bullmq_job_duration_seconds',
  help: 'Job processing duration in seconds',
  labelNames: ['queue', 'job_name', 'worker_id', 'status'] as const,
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 2, 5, 10, 30, 60],
});

const jobsProcessedCounter = new client.Counter({
  name: 'bullmq_jobs_processed_total',
  help: 'Total number of jobs processed',
  labelNames: ['queue', 'job_name', 'worker_id', 'status'] as const,
});

const activeJobsGauge = new client.Gauge({
  name: 'bullmq_active_jobs',
  help: 'Jobs currently being processed by this worker',
  labelNames: ['queue', 'worker_id'] as const,
});

const workerConcurrencyGauge = new client.Gauge({
  name: 'bullmq_worker_concurrency',
  help: 'Current concurrency setting',
  labelNames: ['queue', 'worker_id'] as const,
});

// --- Factory function ---
function createInstrumentedWorker(
  queueName: string,
  processor: (job: Job) => Promise<unknown>,
  connection: Redis,
  workerId = os.hostname()
): { worker: Worker; app: express.Application } {
  const worker = new Worker(
    queueName,
    async (job) => {
      const start = process.hrtime.bigint();
      activeJobsGauge.inc({ queue: queueName, worker_id: workerId });

      try {
        const result = await processor(job);
        const elapsed = Number(process.hrtime.bigint() - start) / 1e9;

        jobDurationHistogram.observe(
          { queue: queueName, job_name: job.name, worker_id: workerId, status: 'completed' },
          elapsed
        );
        jobsProcessedCounter.inc({
          queue: queueName, job_name: job.name, worker_id: workerId, status: 'completed',
        });
        return result;
      } catch (err) {
        const elapsed = Number(process.hrtime.bigint() - start) / 1e9;
        jobDurationHistogram.observe(
          { queue: queueName, job_name: job.name, worker_id: workerId, status: 'failed' },
          elapsed
        );
        jobsProcessedCounter.inc({
          queue: queueName, job_name: job.name, worker_id: workerId, status: 'failed',
        });
        throw err;
      } finally {
        activeJobsGauge.dec({ queue: queueName, worker_id: workerId });
      }
    },
    { connection }
  );

  workerConcurrencyGauge.set(
    { queue: queueName, worker_id: workerId },
    worker.concurrency
  );

  // Expose /metrics endpoint
  const app = express();
  app.get('/metrics', async (_req, res) => {
    res.set('Content-Type', client.register.contentType);
    res.end(await client.register.metrics());
  });

  return { worker, app };
}

// --- Usage ---
const connection = new Redis({
  host: process.env.REDIS_HOST,
  maxRetriesPerRequest: null,
});
const { worker, app } = createInstrumentedWorker(
  'orders',
  async (job) => { /* your processor */ },
  connection,
  `worker-${process.pid}`
);
app.listen(9090, () =>
  console.log('Worker metrics at http://localhost:9090/metrics')
);

OpenTelemetry Custom Metrics with bullmq-otel

If your organization already uses OpenTelemetry, the bullmq-otel package integrates BullMQ metrics into your existing OTel pipeline. You can layer custom worker-level gauges on top:

import { BullMQOtel } from 'bullmq-otel';
import { Worker } from 'bullmq';
import { metrics } from '@opentelemetry/api';

const telemetry = new BullMQOtel({
  tracerName: 'queue-worker',
  meterName: 'queue-worker',
  version: '1.0.0',
  enableMetrics: true,
});

const meter = metrics.getMeter('queue-worker');
const workerSaturation = meter.createGauge('bullmq.worker.saturation', {
  description: 'Ratio of active jobs to max concurrency',
  unit: '1',
});

const worker = new Worker('orders', async (job) => { /* ... */ }, {
  connection,
  telemetry,
  concurrency: 10,
});

setInterval(() => {
  const activeCount = /* track via worker 'active'/'completed' event counts */;
  workerSaturation.record(activeCount / worker.concurrency, {
    'worker.id': os.hostname(),
    'queue.name': 'orders',
  });
}, 5000);

Pro tip: Label every metric by worker_id (hostname, pod name, or PID). When you have 20 workers and one is performing differently, per-worker labels are the difference between "p99 is high" and "worker-7 has a memory leak."

Worker Health Probes — Liveness and Readiness That Reflect Real Capacity

Kubernetes health checks are table stakes, but a simple TCP probe that returns 200 when the process is alive doesn't tell you whether the worker can actually process jobs. The process can be running but have a dead Redis connection, be paused, or be stuck on a stalled job.

A meaningful worker health check must reflect real processing capacity.

Production Health Check Server

This class integrates directly with the BullMQ worker's event system to serve three probe endpoints for Kubernetes:

import { Worker } from 'bullmq';
import { Redis } from 'ioredis';
import express from 'express';

interface HealthConfig {
  maxStaleSeconds: number;      // max seconds without completing a job
  maxEventLoopLagMs: number;    // max event loop lag before overloaded
  maxMemoryPercent: number;     // max heap usage %
  startupGracePeriodMs: number; // grace window for initial startup
}

class WorkerHealthServer {
  private app: express.Application;
  private worker: Worker;
  private connection: Redis;
  private config: HealthConfig;
  private lastCompletedAt: number = Date.now();
  private errors: string[] = [];
  private isStarted = false;
  private isShuttingDown = false;

  constructor(
    worker: Worker,
    connection: Redis,
    config?: Partial<HealthConfig>
  ) {
    this.worker = worker;
    this.connection = connection;
    this.config = {
      maxStaleSeconds: config?.maxStaleSeconds ?? 120,
      maxEventLoopLagMs: config?.maxEventLoopLagMs ?? 100,
      maxMemoryPercent: config?.maxMemoryPercent ?? 90,
      startupGracePeriodMs: config?.startupGracePeriodMs ?? 10_000,
    };
    this.app = express();

    worker.on('completed', () => { this.lastCompletedAt = Date.now(); });
    worker.on('ready', () => { this.isStarted = true; });
    worker.on('error', (err) => {
      this.errors.push(err.message);
      if (this.errors.length > 100) this.errors.shift();
    });

    this.registerRoutes();
  }

  private registerRoutes(): void {
    // Startup probe — returns 200 once initialized
    this.app.get('/health/startup', (_req, res) => {
      if (this.isStarted) return res.json({ status: 'ok' });
      if (Date.now() - process.uptime() * 1000 > this.config.startupGracePeriodMs) {
        return res.json({ status: 'ok' });
      }
      res.status(503).json({ status: 'not_ready', message: 'Initializing' });
    });

    // Liveness probe — process is alive and not stuck
    this.app.get('/health/live', async (_req, res) => {
      if (this.isShuttingDown) {
        return res.status(503).json({ status: 'shutting_down' });
      }

      // Redis connectivity check
      try {
        await this.connection.ping();
      } catch {
        return res.status(503).json({ status: 'unhealthy', reason: 'redis_unreachable' });
      }

      // Stalled worker detection: jobs waiting but nothing completing
      const idleTime = Date.now() - this.lastCompletedAt;
      if (idleTime > this.config.maxStaleSeconds * 1000) {
        try {
          const counts = await this.worker.getWaitingCount();
          if (counts > 0) {
            return res.status(503).json({
              status: 'stuck',
              reason: `No jobs completed in ${Math.round(idleTime / 1000)}s, ${counts} waiting`,
            });
          }
        } catch {
          // getWaitingCount may fail if Redis is slow — treat as degraded
        }
      }

      res.json({ status: 'ok', uptime: Math.floor(process.uptime()) });
    });

    // Readiness probe — can accept new jobs
    this.app.get('/health/ready', async (_req, res) => {
      if (this.isShuttingDown) return res.status(503).json({ status: 'shutting_down' });
      if (this.worker.isPaused()) return res.status(503).json({ status: 'paused' });

      // Check event loop lag
      const lag = await this.measureEventLoopLag();
      if (lag > this.config.maxEventLoopLagMs) {
        return res.status(503).json({
          status: 'overloaded',
          reason: `event_loop_lag:${lag}ms`,
        });
      }

      // Check memory pressure
      const memUsage = process.memoryUsage();
      const memPercent = (memUsage.heapUsed / memUsage.heapTotal) * 100;
      if (memPercent > this.config.maxMemoryPercent) {
        return res.status(503).json({
          status: 'overloaded',
          reason: `memory:${memPercent.toFixed(0)}%`,
        });
      }

      res.json({
        status: 'ok',
        checks: {
          event_loop_lag_ms: lag,
          memory_percent: memPercent.toFixed(1),
          concurrency: this.worker.concurrency,
          is_running: this.worker.isRunning(),
        },
      });
    });
  }

  private measureEventLoopLag(): Promise<number> {
    return new Promise((resolve) => {
      const start = Date.now();
      setImmediate(() => resolve(Date.now() - start));
    });
  }

  get appInstance(): express.Application {
    return this.app;
  }

  startShutdown(): void {
    this.isShuttingDown = true;
  }
}

Pro tip: Combine the readiness check with worker.pause() during Kubernetes preStop hooks. Make the readiness endpoint return 503, then call worker.pause(true) — this tells the service mesh to stop routing traffic before the worker stops accepting jobs, eliminating the window where the orchestrator sends a SIGTERM mid-job. The sequence: readiness fails -> pause -> drain in-flight -> SIGTERM arrives -> close.

Advanced Lifecycle Management — Pause, Drain, Resume and Graceful Reconfiguration

Beyond worker.close(), BullMQ provides a full lifecycle API that gives you fine-grained control over worker state transitions. These are essential for zero-downtime operations.

The Lifecycle API at a Glance

Method What It Does When to Use
worker.pause(waitActive) Stops pulling new jobs. Active jobs finish if waitActive=true Before reconfiguring, during maintenance windows
worker.resume() Resumes pulling jobs after a pause After maintenance, after reconfigure
worker.drain() Removes all waiting jobs from the queue without processing them Before a migration, canceling scheduled downtime
worker.isRunning() Returns whether the worker is processing jobs Health checks, status endpoints
worker.isPaused() Returns whether the worker is paused Health checks

Full Lifecycle Manager

This class wraps the complete lifecycle into a single managed abstraction with safe reconfiguration, drain, and graceful shutdown:

import { Worker, Queue, Job } from 'bullmq';

type WorkerConfig = {
  concurrency: number;
  lockDuration?: number;
};

class ManagedWorker {
  private worker: Worker | null = null;
  private queueName: string;
  private processor: (job: Job) => Promise<unknown>;
  private connection: Redis;
  private config: WorkerConfig;

  constructor(
    queueName: string,
    processor: (job: Job) => Promise<unknown>,
    connection: Redis,
    config: WorkerConfig
  ) {
    this.queueName = queueName;
    this.processor = processor;
    this.connection = connection;
    this.config = config;
  }

  start(): boolean {
    if (this.worker?.isRunning()) return false;
    this.worker = new Worker(this.queueName, this.processor, {
      connection: this.connection,
      concurrency: this.config.concurrency,
      lockDuration: this.config.lockDuration,
    });
    return true;
  }

  async pause(doNotWaitActive = false): Promise<void> {
    await this.assertRunning();
    await this.worker!.pause(doNotWaitActive);
  }

  async resume(): Promise<void> {
    await this.assertRunning();
    await this.worker!.resume();
  }

  async setConcurrency(n: number): Promise<void> {
    await this.assertRunning();
    this.worker!.concurrency = n;
    this.config.concurrency = n;
  }

  async reconfigure(
    newConfig: WorkerConfig,
    drainTimeoutMs = 30_000
  ): Promise<void> {
    if (this.worker?.isRunning()) {
      await this.worker.pause(true);
    }

    await this.drainQueue(drainTimeoutMs);
    this.config = { ...this.config, ...newConfig };

    if (this.worker) {
      this.worker.concurrency = this.config.concurrency;
      await this.worker.resume();
    } else {
      this.start();
    }
  }

  async drainQueue(timeoutMs = 10_000): Promise<number> {
    const queue = new Queue(this.queueName, { connection: this.connection });
    try {
      const before = await queue.getJobCounts();
      const waiting = (before.waiting ?? 0) + (before.delayed ?? 0);
      if (waiting === 0) return 0;
      await queue.drain();
      console.log(`[ManagedWorker] drained ${waiting} jobs`);
      return waiting;
    } finally {
      await queue.close();
    }
  }

  getStatus() {
    return {
      isRunning: this.worker?.isRunning() ?? false,
      isPaused: this.worker?.isPaused() ?? false,
      concurrency: this.worker?.concurrency ?? 0,
    };
  }

  async shutdown(timeoutMs = 30_000): Promise<void> {
    if (!this.worker?.isRunning()) return;
    await this.pause(true);

    const closePromise = this.worker.close();
    const timeoutPromise = new Promise<void>((_, reject) =>
      setTimeout(() => reject(new Error('Shutdown timed out')), timeoutMs)
    );

    await Promise.race([closePromise, timeoutPromise]).catch(async (err) => {
      console.error('[ManagedWorker] force-close after timeout:', err.message);
      this.worker?.removeAllListeners();
      await this.worker?.close().catch(() => {});
    });
  }

  private async assertRunning(): Promise<void> {
    if (!this.worker?.isRunning()) {
      throw new Error('Worker not running. Call start() first.');
    }
  }
}

With this manager, you can hot-reconfigure workers at runtime via signal handlers:

const mw = new ManagedWorker('email', async (job) => {
  await sendEmail(job.data);
}, connection, { concurrency: 10 });

mw.start();

// Reconfigure on SIGUSR2 — no deploy needed
process.on('SIGUSR2', async () => {
  console.log('Hot reconfiguring...');
  await mw.reconfigure({ concurrency: 5, lockDuration: 60_000 });
  console.log('Reconfiguration applied');
});

Pro tip: worker.pause(true) is not instantaneous — it waits for all active jobs to complete before resolving. When using it in a Kubernetes preStop hook, set terminationGracePeriodSeconds to your longest expected job duration plus 10 seconds. If a job takes 5 minutes, the grace period must exceed 5 minutes or Kubernetes will SIGKILL the pod mid-job.

Putting It All Together — A Production Worker Daemon

Each pattern above solves one problem. In production, you need them all working together: multi-queue routing for job dispatch, adaptive concurrency for efficiency, Prometheus metrics for observability, health probes for Kubernetes, and lifecycle management for operational control.

Here's how they compose into a single deployable worker daemon:

import { Worker, Job } from 'bullmq';
import { Redis } from 'ioredis';
import express from 'express';
import os from 'os';

async function main() {
  const connection = new Redis({
    host: process.env.REDIS_HOST ?? 'localhost',
    port: Number(process.env.REDIS_PORT ?? 6379),
    maxRetriesPerRequest: null,
    retryStrategy: (times) => Math.min(times * 100, 5000),
  });

  const workerId = process.env.HOSTNAME ?? `worker-${process.pid}`;

  // 1. Create instrumented worker (custom metrics from Section 3)
  const { worker, app } = createInstrumentedWorker(
    'shared-work',
    routedProcessor,  // dispatch-by-name from Section 1
    connection,
    workerId
  );

  // 2. Attach adaptive concurrency (Section 2)
  //    -- setInterval monitors CPU/memory, adjusts worker.concurrency

  // 3. Health probes on the same HTTP server (Section 4)
  const healthServer = new WorkerHealthServer(worker, connection);
  app.use('/health', healthServer.appInstance);
  app.listen(9090);

  // 4. Signal-based lifecycle management (Section 5)
  process.on('SIGTERM', async () => {
    console.log('[daemon] SIGTERM received, shutting down...');
    healthServer.startShutdown();
    // worker.pause(true) is called implicitly by worker.close()
    await worker.close();
    await connection.quit();
    process.exit(0);
  });

  console.log(`[daemon] Worker ${workerId} started on shared-work queue`);
}

main().catch((err) => {
  console.error('Fatal error:', err);
  process.exit(1);
});

The retryStrategy on the ioredis connection is critical. Without it, ioredis retries immediately and aggressively, flooding Redis with reconnection attempts during an outage and making recovery worse. Cap it at 10-20 seconds.

Recap

  • Multi-queue workers let you consolidate processing logic, save Redis connections, and route by job name or source queue. Use the shared-queue-with-names pattern for simplicity; use the coordinator pattern when queues have different SLAs.
  • Dynamic concurrency turns a static knob into a live control. Use CPU/memory gauges for auto-tuning and API status codes for demand-aware throttling.
  • Custom instrumentation moves you from "is the queue healthy?" to "is this worker healthy?" Label metrics by worker ID, job type, and environment for actionable dashboards.
  • Health probes that reflect actual processing capacity prevent Kubernetes from routing traffic to dead workers and catch stuck workers before jobs pile up.
  • Lifecycle management beyond close() — using pause(), resume(), drain(), and reconfiguration — gives you zero-downtime operational control over your workers.

For more production BullMQ patterns, check out our other guides on rate limiting, OpenTelemetry tracing for workers, and Redis memory management.

Try QueueHub for real-time visibility into your BullMQ workers — see per-worker metrics, job lifecycle tracking, and Redis health in one dashboard. Free tier available for small teams.

← All Articles
Share on XShare on LinkedIn

Related Articles

2026-07-12

Best BullMQ Dashboard Alternatives in 2026: A Comprehensive Comparison

Comparing every BullMQ UI option side by side: Bull Board, Arena, Taskforce, QueueHub, and raw redis-cli. Feature matrices, pricing, pros and cons, and recommendations for every team size.

2026-07-10

QueueHub vs pg-boss: Redis vs PostgreSQL as a Job Queue Backend

pg-boss uses PostgreSQL as a Node.js job queue, while BullMQ (monitored by QueueHub) uses Redis. We compare the two approaches across throughput, persistence, transactional queues, deployment, and when transactional enqueueing matters.

2026-07-09

QueueHub vs Temporal: Job Queues vs Workflow Orchestration

Temporal is a workflow orchestration platform — a different category from Redis-backed job queues. We compare Temporal.io against BullMQ + QueueHub, covering complexity, use cases, durability, observability, and when to choose each.