Usage - BullMQ

Quick Start

Add Jobs to a Queue

The simplest way to add jobs is using the stateless API:

# Add a job
{:ok, job} = BullMQ.Queue.add("emails", "send-welcome", %{
  to: "user@example.com",
  template: "welcome"
}, connection: :my_redis)

# Add a delayed job (delay in milliseconds)
{:ok, job} = BullMQ.Queue.add("emails", "reminder", %{message: "Hello!"},
  connection: :my_redis,
  delay: 60_000  # 1 minute
)

# Add a prioritized job (lower number = higher priority)
{:ok, job} = BullMQ.Queue.add("emails", "urgent", %{},
  connection: :my_redis,
  priority: 1
)

Process Jobs with a Worker

Create a processor function and start a worker:

defmodule MyApp.EmailWorker do
  def process(%BullMQ.Job{name: "send-welcome", data: data}) do
    MyApp.Mailer.send_welcome(data["to"], data["template"])
    {:ok, %{sent: true}}
  end

  def process(%BullMQ.Job{name: name}) do
    {:error, "Unknown job type: #{name}"}
  end
end

# Start a worker
{:ok, worker} = BullMQ.Worker.start_link(
  queue: "emails",
  connection: :my_redis,
  processor: &MyApp.EmailWorker.process/1,
  concurrency: 5
)

Adding to Supervision Tree

For production use, add workers to your application’s supervision tree:

defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      # Redis connection
      {Redix, name: :my_redis, host: "localhost", port: 6379},

      # BullMQ Worker
      {BullMQ.Worker,
        queue: "emails",
        connection: :my_redis,
        processor: &MyApp.EmailWorker.process/1,
        concurrency: 5
      }
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

Job Options

Basic Options

BullMQ.Queue.add("tasks", "process-data", %{data: "..."},
  connection: :my_redis,
  job_id: "custom-id-123",      # Custom job ID
  priority: 1,                   # Lower = higher priority
  delay: 60_000,                 # Delay 60 seconds
  attempts: 5,                   # Retry up to 5 times
  backoff: %{
    type: "exponential",
    delay: 1000
  },
  remove_on_complete: true,      # Auto-remove on success
  remove_on_fail: 100            # Keep last 100 failed jobs
)

Job Deduplication

# Prevent duplicate jobs
BullMQ.Queue.add("tasks", "send-notification", %{user_id: 123},
  connection: :my_redis,
  deduplication: %{
    id: "notify-user-123",
    ttl: 60_000  # Deduplicate for 60 seconds
  }
)

Worker Configuration

Worker with Event Callbacks

{:ok, worker} = BullMQ.Worker.start_link(
  queue: "tasks",
  connection: :my_redis,
  processor: &process/1,
  concurrency: 10,
  on_completed: fn job, result ->
    IO.puts("Job #{job.id} completed: #{inspect(result)}")
  end,
  on_failed: fn job, reason ->
    IO.puts("Job #{job.id} failed: #{reason}")
  end,
  on_active: fn job ->
    IO.puts("Job #{job.id} started")
  end,
  on_stalled: fn job_id ->
    IO.puts("Job #{job_id} stalled")
  end
)

Rate Limiting

{:ok, worker} = BullMQ.Worker.start_link(
  queue: "api-calls",
  connection: :my_redis,
  processor: &process/1,
  limiter: %{
    max: 100,        # Max jobs
    duration: 60_000  # Per 60 seconds (100 jobs/minute)
  }
)

Job Processing

Basic Processor

def process(%BullMQ.Job{data: data} = job) do
  # Do work
  result = do_some_work(data)
  
  # Return success
  {:ok, result}
end

With Progress Updates

def process(%BullMQ.Job{} = job) do
  total = 100
  
  Enum.each(1..total, fn i ->
    do_work_step(i)
    
    # Update progress (0-100)
    BullMQ.Worker.update_progress(job, i)
  end)
  
  {:ok, "completed"}
end

Error Handling

def process(%BullMQ.Job{data: data}) do
  case risky_operation(data) do
    {:ok, result} -> {:ok, result}
    {:error, :temporary} -> {:error, "Temporary failure"}  # Will retry
    {:error, :invalid} -> {:error, :unrecoverable}  # Won't retry
  end
end

Queue Events

Subscribe to queue-level events using QueueEvents:

# Start QueueEvents
{:ok, events} = BullMQ.QueueEvents.start_link(
  queue: "tasks",
  connection: :my_redis
)

# Subscribe to events
BullMQ.QueueEvents.subscribe(events)

# Handle events
receive do
  {:bullmq_event, :completed, %{"jobId" => id}} ->
    IO.puts("Job #{id} completed!")
  
  {:bullmq_event, :failed, %{"jobId" => id, "failedReason" => reason}} ->
    IO.puts("Job #{id} failed: #{reason}")
  
  {:bullmq_event, :progress, %{"jobId" => id, "data" => progress}} ->
    IO.puts("Job #{id} progress: #{progress}%")
end

Job Schedulers (Repeatable Jobs)

Cron-based Scheduling

# Run every hour
{:ok, job} = BullMQ.JobScheduler.upsert(
  :my_redis,
  "maintenance",
  "cleanup",
  %{pattern: "0 * * * *"},  # Cron pattern
  "cleanup-job",
  %{type: "hourly"},
  prefix: "bull"
)

Interval-based Scheduling

# Run every minute
{:ok, job} = BullMQ.JobScheduler.upsert(
  :my_redis,
  "heartbeats",
  "ping",
  %{every: 60_000},  # Every 60 seconds
  "heartbeat",
  %{},
  prefix: "bull"
)

Managing Schedulers

# List all schedulers
{:ok, schedulers} = BullMQ.JobScheduler.list(
  :my_redis,
  "maintenance",
  prefix: "bull"
)

# Remove a scheduler
{:ok, removed} = BullMQ.JobScheduler.remove(
  :my_redis,
  "maintenance",
  "cleanup",
  prefix: "bull"
)

Queue Operations

Get Queue Information

# Get job counts
{:ok, counts} = BullMQ.Queue.get_counts("emails", connection: :my_redis)
# => %{waiting: 10, active: 2, delayed: 5, completed: 100, failed: 3, ...}

# Get jobs in a specific state
{:ok, jobs} = BullMQ.Queue.get_jobs("emails", [:waiting, :delayed],
  connection: :my_redis,
  start: 0,
  end: 9
)

# Get a specific job
{:ok, job} = BullMQ.Queue.get_job("emails", "job-id-123",
  connection: :my_redis
)

# Get job state
{:ok, state} = BullMQ.Queue.get_job_state("emails", "job-id-123",
  connection: :my_redis
)
# => :waiting | :active | :delayed | :completed | :failed

Pause and Resume

# Pause the queue
:ok = BullMQ.Queue.pause("emails", connection: :my_redis)

# Resume the queue
:ok = BullMQ.Queue.resume("emails", connection: :my_redis)

# Check if paused
{:ok, is_paused} = BullMQ.Queue.paused?("emails", connection: :my_redis)

Clean Up Jobs

# Drain the queue (remove all waiting jobs)
:ok = BullMQ.Queue.drain("emails", connection: :my_redis)

# Remove a specific job
:ok = BullMQ.Queue.remove_job("emails", "job-id-123",
  connection: :my_redis
)

# Retry a failed job
:ok = BullMQ.Queue.retry_job("emails", "job-id-123",
  connection: :my_redis
)

Parent-Child Jobs (Flows)

Create job dependencies using FlowProducer:

{:ok, flow} = BullMQ.FlowProducer.start_link(
  connection: :my_redis,
  prefix: "bull"
)

# Create a flow with parent and children
{:ok, flow_job} = BullMQ.FlowProducer.add(flow, %{
  name: "process-order",
  queue_name: "orders",
  data: %{order_id: 123},
  children: [
    %{
      name: "send-email",
      queue_name: "emails",
      data: %{order_id: 123}
    },
    %{
      name: "update-inventory",
      queue_name: "inventory",
      data: %{order_id: 123}
    }
  ]
})

Graceful Shutdown

# Close worker and wait for active jobs to finish
:ok = BullMQ.Worker.close(worker)

# Force close without waiting
:ok = BullMQ.Worker.close(worker, force: true)

Interoperability

Jobs added in Elixir can be processed by workers in other languages: Node.js Worker:

import { Worker } from 'bullmq';

const worker = new Worker('emails', async job => {
  console.log('Processing:', job.data);
  return { success: true };
});

Python Worker:

from bullmq import Worker

async def process(job, job_token):
    print(f"Processing: {job.data}")
    return {"success": True}

worker = Worker("emails", process)

Best Practices

Use supervision trees - Always supervise workers for automatic restarts
Set appropriate concurrency - Balance between throughput and resource usage
Use named connections - Makes it easier to manage multiple Redis instances
Handle errors gracefully - Return {:error, reason} for retries, :error or raise for failures
Monitor with Telemetry - Set up telemetry handlers for observability
Use rate limiting - Prevent overwhelming external services
Clean up old jobs - Use remove_on_complete and remove_on_fail options

View Changelog

See what’s new in the latest version

​Quick Start

​Add Jobs to a Queue

​Process Jobs with a Worker

​Adding to Supervision Tree

​Job Options

​Basic Options

​Job Deduplication

​Worker Configuration

​Worker with Event Callbacks

​Rate Limiting

​Job Processing

​Basic Processor

​With Progress Updates

​Error Handling

​Queue Events

​Job Schedulers (Repeatable Jobs)

​Cron-based Scheduling

​Interval-based Scheduling

​Managing Schedulers

​Queue Operations

​Get Queue Information

​Pause and Resume

​Clean Up Jobs

​Parent-Child Jobs (Flows)

​Graceful Shutdown

​Interoperability

​Best Practices