DOCSEARCHbyAlgolia
Submit Docs

Exq Documentation

repository·master·Indexed 23 days ago

https://github.com/akira/exq

A job processing library for Elixir compatible with Resque and Sidekiq. Exq uses Redis as a backend to provide durable, persistent, and retriable background jobs, featuring support for exponential backoff, dead job queues, unique jobs, and custom middleware.

Tokens
7.4K
Snippets
19
Records
53
Agent score
81%

What's inside Exq

  1. Configure Node Recovery for unexpected shutdowns

    master

    Exq provides two mechanisms to recover jobs left in an 'in-progress' state after a node crash:

    1. Same Node Recovery

    Exq uses a node_identifier to track jobs. By default, this is the machine's hostname. If a node restarts with the same identifier, it re-enqueues its in-progress jobs. In environments like Kubernetes where hostnames change, implement a custom Exq.NodeIdentifier.Behaviour.

    2. Heartbeat Mechanism

    If a node is gone for a long time (e.g., autoscaling), the heartbeat mechanism detects it. If a node misses missed_heartbeats_allowed consecutive heartbeats, its in-progress jobs are re-enqueued. This is disabled by default.

    Configuration:

    config :exq,
    heartbeat_enable: true,
    heartbeat_interval: 60_000,
    missed_heartbeats_allowed: 5
    # Custom Node Identifier implementation
defmodule MyApp.CustomNodeIdentifier do
  @behaviour Exq.NodeIdentifier.Behaviour

  def node_id do
     System.get_env("NODE_ID")
  end
end

# In config.exs
config :exq,
   node_identifier: MyApp.CustomNodeIdentifier

  2. Implement custom middleware

    master

    Exq supports custom middleware to customize worker execution. To use it, add your module to the middleware list in your configuration.

    Your middleware module must implement the following behaviors:

    • before_work/1
    • after_processed_work/1
    • after_failed_work/1

    Note: You can halt the execution chain within these functions.

  3. Use Unique Jobs to prevent duplicates

    master

    Exq provides a best-effort mechanism to avoid duplicate jobs using a lock abstraction. This requires the Exq.Middleware.Unique middleware to be present in your configuration.

    How it works: A lock is created based on the queue, class, and arguments (or a provided unique_token). If a duplicate is enqueued before the lock expires, Exq.enqueue/5 returns {:conflict, jid}.

    Key Options:

    • unique_for: (Mandatory) The maximum duration (seconds) the lock remains active.
    • unique_until: Controls when the lock is cleared:
      • :success: Clears on successful completion or if the job dies.
      • :start: Clears when the job is first picked up for execution.
      • :expiry: Clears based on the unique_for duration.

    Common Patterns:

    • Idempotency: Use unique_until: :expiry to cover retry periods.
    • Debounce: Use unique_until: :start to ensure only one job is pending in the queue for a specific set of args.
    • Batching: Use enqueue_in with unique_until: :success to prevent new jobs from being enqueued until the scheduled batch job completes.

  4. Snooze a job using `Exq.Middleware.Snooze`

    master

    A worker can delay its next execution by returning {:snooze, seconds} from its perform/1 method.

    To enable this, you must add Exq.Middleware.Snooze to your Exq middleware configuration. If you are also using Exq.Middleware.Unique, ensure Exq.Middleware.Snooze is placed before it in the list.

    defmodule MyWorker do
  def perform do
    {:snooze, 10}
  end
end

  5. Test Exq workers with `Exq.Mock`

    master

    Use Exq.Mock to test workers without requiring a real Redis instance.

    1. Set the queue_adapter to Exq.Adapters.Queue.Mock in your config/test.exs.
    2. Start the mock server in test_helper.exs using Exq.Mock.start_link(mode: mode).

    Supported modes:

    • :redis: Used for integration testing. Does not support the async: true option.
    • :fake: Jobs are enqueued in a local queue and never executed. Exq.Mock.jobs() can be used to inspect them. Supports async: true.
    • :inline: Jobs are executed immediately in the same process. Supports async: true.

    You can switch modes during a test using Exq.Mock.set_mode(:fake).

    # In config/test.exs
config :exq,
  queue_adapter: Exq.Adapters.Queue.Mock

# In test_helper.exs
Exq.Mock.start_link(mode: :redis)

  7. Integrate Exq into an OTP application

    master

    You can include :exq in your application's supervision tree. If you add :exq to your applications list in def application, it will start automatically using settings from config.exs and register under the name Elixir.Exq.

    Manual Supervision (Recommended for Phoenix/Ecto): If you need to control the order of startup (e.g., ensuring your database repository starts before Exq workers), set start_on_application: false in your config and add Exq manually to your supervision tree.

    # In config.exs
config :exq,
   start_on_application: false

# In your application supervisor
def start(_type, _args) do
  children = [
    MyApp.Repo,
    MyApp.Endpoint,
    Exq,
  ]
  Supervisor.start_link(children, [])
end

  8. Install Exq via mix

    master

    To install Exq, add :exq to your mix.exs dependencies. Replace the version with the latest available on Hex.pm.

    defp deps do
  [
    # ... other deps
    {:exq, "~> 0.24.0"}
  ]
end

    After updating mix.exs, run mix deps.get to fetch the dependency.

  9. Implement a custom Exq middleware

    master

    To create a custom middleware, define a module that implements the Exq.Middleware.Behaviour. The middleware must provide three functions: before_work/1, after_processed_work/1, and after_failed_work/1. Each function receives an Exq.Middlewares.Pipeline structure and must return the same structure (modified or unmodified).

    • before_work/1: Used to update worker state before a job is processed.
    • after_processed_work/1: Used for cleanup or notifications after a successful job.
    • after_failed_work/1: Used for cleanup or notifications after a job fails.
    defmodule MyMiddleware do
  @behaviour Exq.Middleware.Behaviour

  def before_work(pipeline) do
    # some functionality goes here...
    pipeline
  end

  def after_processed_work(pipeline) do
    # some functionality goes here...
    pipeline
  end

  def after_failed_work(pipeline) do
    # some functionality goes here...
    pipeline
  end
end

  10. Configure Exq queues and concurrency

    master

    You can specify which queues Exq listens to using the queues option. Concurrency (the number of concurrent workers) can be set globally for all queues or defined on a per-queue basis.

    Global Concurrency:

    config :exq,
  concurrency: 1000,
  queues: ["default"]

    Per-Queue Concurrency: Pass a list of tuples in the format {"queue_name", concurrency_limit} to the queues option.

    config :exq,
  queues: [{"q1", 10_000}, {"q2", 10}]

  11. Configure job retries and dead jobs

    master

    Exq supports automatic retries with exponential backoff. To enable this, set scheduler_enable to true and max_retries to a value greater than 0.

    Jobs that exceed max_retries are moved to a "dead jobs" queue. You can configure the maximum number of dead jobs to keep and how long they should persist using dead_max_jobs and dead_timeout_in_seconds.

    config :exq,
  scheduler_enable: true,
  max_retries: 25,
  dead_max_jobs: 10_000,
  dead_timeout_in_seconds: 180 * 24 * 60 * 60

  12. Configure Exq startup modes

    master

    The mode option controls which Exq components are started. This is useful for separating nodes that only enqueue jobs from nodes that actually run workers.

    Available modes:

    • :default: Starts worker, enqueuer, and API.
    • :enqueuer: Starts only the enqueuer.
    • :api: Starts only the API.
    • [:api, :enqueuer]: Starts both the enqueuer and API.