Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/msolli/proletarian/llms.txt

Use this file to discover all available pages before exploring further.

A queue worker is the engine at the heart of Proletarian. It is a long-running process that continuously polls a database table for jobs, hands each job off to your handler function, and then archives the completed job — all within a database transaction. You create one with proletarian.worker/create-queue-worker, passing it a javax.sql.DataSource and your handler function, then start it with worker/start!.

​
The Poll/Run Cycle

Each worker thread operates in a tight loop:
  1. Poll — the thread queries the job table for the next available job whose process_at time is in the past.
  2. Lock — the row is locked using SELECT … FOR UPDATE SKIP LOCKED, so no other thread or machine can pick it up.
  3. Run — your handler function is called with the job’s type and payload.
  4. Archive — on success, the job is deleted from the queue table and inserted into the archived_job table with a :success status, both in the same transaction.
  5. Repeat — the thread immediately polls for another job. When the queue is empty, it waits for :proletarian/polling-interval-ms before polling again.
If the handler throws an exception the job is not archived; instead Proletarian invokes the retry strategy.

​
Thread Pool and Parallelism

The :proletarian/worker-threads option controls how many worker threads run inside one JVM process. Each thread independently polls the queue and runs jobs, so a worker with worker-threads 4 can process four jobs simultaneously. 
(require '[proletarian.worker :as worker])

(def my-worker
  (worker/create-queue-worker
    data-source
    handle-job!
    {:proletarian/worker-threads      4
     :proletarian/polling-interval-ms 500}))

​
Scaling Across Multiple Machines

A queue worker is local to a single JVM process. When you run Proletarian on multiple machines (e.g. in a horizontally scaled deployment), each machine runs its own worker process. The total parallelism for a given queue is: 
total parallelism = number of machines × worker-threads per machine
Because every worker uses SKIP LOCKED, they can all poll the same queue table concurrently without lock contention — each thread atomically claims a row that no other thread has touched.

​
Default Queue and Named Queues

If you don’t specify a queue, both job/enqueue! and worker/create-queue-worker use the default queue :proletarian/default. All jobs live in the same database table and are differentiated by the queue column. Named queues let you tune throughput and priority independently: 
;; High-priority queue with more threads and faster polling
(def email-worker
  (worker/create-queue-worker
    data-source
    handle-email-job!
    {:proletarian/queue               :email
     :proletarian/worker-threads      8
     :proletarian/polling-interval-ms 100}))

;; Low-priority background queue
(def report-worker
  (worker/create-queue-worker
    data-source
    handle-report-job!
    {:proletarian/queue               :reports
     :proletarian/worker-threads      1
     :proletarian/polling-interval-ms 5000}))

​
Creating, Starting, and Stopping a Worker

(ns your-app.workers
  (:require [next.jdbc :as jdbc]
            [proletarian.worker :as worker]
            [your-app.handlers :as handlers]))

(def data-source
  (jdbc/get-datasource "jdbc:postgresql://localhost/myapp"))

;; Create the worker (does not start polling yet)
(def email-worker
  (worker/create-queue-worker
    data-source
    handlers/handle-job!
    {:proletarian/queue                    :email
     :proletarian/worker-threads           2
     :proletarian/polling-interval-ms      200
     :proletarian/install-jvm-shutdown-hook? true}))

;; Start polling
(worker/start! email-worker)

;; Stop polling gracefully (waits for in-flight jobs to finish)
(worker/stop! email-worker)
In production, set :proletarian/install-jvm-shutdown-hook? to true. Proletarian will register a JVM shutdown hook that calls stop! automatically when the process receives a SIGTERM or SIGINT, giving in-flight jobs a chance to complete before the JVM exits. The default timeout for waiting on in-flight jobs is 10 seconds, configurable via :proletarian/await-termination-timeout-ms.

​
Identifying Workers with queue-worker-id

The :proletarian/queue-worker-id option sets a string identifier for the worker. It is used as the thread-name prefix for all threads in the worker’s pool, making it easy to identify worker threads in stack traces and thread dumps. It is also included in every log event under the :proletarian.worker/queue-worker-id key. 
(worker/create-queue-worker
  data-source
  handle-job!
  {:proletarian/queue-worker-id "email-worker-us-east-1"})
If you omit this option, Proletarian derives a default from the queue name: proletarian[:email].

​
The on-shutdown Callback

The :proletarian/on-shutdown option accepts a zero-argument function that Proletarian calls after the worker has fully shut down. Use it to release resources, emit a final log line, or signal a health-check system. 
(worker/create-queue-worker
  data-source
  handle-job!
  {:proletarian/on-shutdown
   (fn []
     (log/info "Queue worker shut down cleanly."))})

​
SKIP LOCKED Semantics

Proletarian uses SELECT … FOR UPDATE SKIP LOCKED when polling for jobs. This PostgreSQL/MySQL feature means:
  • A thread only sees rows that are not currently locked by another transaction.
  • If two threads poll at the same moment they will each claim a different row — neither blocks waiting for the other.
  • There is no need for application-level coordination between threads or machines; the database enforces mutual exclusion.
SKIP LOCKED requires PostgreSQL 9.5+ or MySQL 8.0.1+. Proletarian enforces these minimum versions as a hard dependency.

Build docs developers (and LLMs) love

Get started for freeTalk to us