Use this file to discover all available pages before exploring further.
Getting Proletarian running involves two steps: adding the library dependency to your project, and creating the two database tables that Proletarian uses to store and archive jobs. Both steps are covered below.
Proletarian relies on the SKIP LOCKED clause when polling for jobs. This allows multiple concurrent worker threads (and workers on multiple machines) to each claim a distinct job from the queue without contention.
Proletarian does not pull in a Clojure database library as a dependency. It works with any javax.sql.DataSource-compatible connection pool or JDBC helper you are already using — next.jdbc, clojure.java.jdbc, or anything else. You pass your existing data source directly to worker/create-queue-worker, and an open java.sql.Connection to job/enqueue!.
Proletarian needs two tables — one for queued jobs and one for archived (completed or failed) jobs — plus an index. Copy the SQL for your database below into a migration file and run it before starting the application.
The index job_queue_process_at on (queue, process_at) is critical for
performance. Without it, every poll operation will perform a full table scan.
Do not omit or drop this index in production.
PostgreSQL
MySQL
The following SQL creates the proletarian schema, both tables, and the required index. It is safe to run multiple times (IF NOT EXISTS guards are included).
CREATE SCHEMA IF NOT EXISTS proletarian;CREATE TABLE IF NOT EXISTS proletarian.job( job_id UUID PRIMARY KEY, -- job id, generated and returned by proletarian.job/enqueue! queue TEXT NOT NULL, -- queue name job_type TEXT NOT NULL, -- job type payload TEXT NOT NULL, -- Transit-encoded job data attempts INTEGER NOT NULL, -- Number of attempts. Starts at 0. Increments when the job is processed. enqueued_at TIMESTAMPTZ NOT NULL, -- When the job was enqueued (never changes) process_at TIMESTAMPTZ NOT NULL -- When the job should be run (updates every retry));CREATE TABLE IF NOT EXISTS proletarian.archived_job( job_id UUID PRIMARY KEY, -- Copied from job record. queue TEXT NOT NULL, -- Copied from job record. job_type TEXT NOT NULL, -- Copied from job record. payload TEXT NOT NULL, -- Copied from job record. attempts INTEGER NOT NULL, -- Copied from job record. enqueued_at TIMESTAMPTZ NOT NULL, -- Copied from job record. process_at TIMESTAMPTZ NOT NULL, -- Copied from job record (data for the last run only) status TEXT NOT NULL, -- success / failure finished_at TIMESTAMPTZ NOT NULL -- When the job was finished (success or failure));DROP INDEX IF EXISTS proletarian.job_queue_process_at;CREATE INDEX job_queue_process_at ON proletarian.job (queue, process_at);
The schema proletarian groups both tables together. The proletarian.job table holds jobs that are waiting to be processed or are currently in-flight. The proletarian.archived_job table is the permanent record of every completed job, stored with a status of success or failure.
The following SQL creates the proletarian database (MySQL uses databases where PostgreSQL uses schemas), both tables, and the required composite index.
CREATE DATABASE IF NOT EXISTS proletarian;CREATE TABLE IF NOT EXISTS proletarian.job ( job_id BINARY(16) PRIMARY KEY, -- job id, generated and returned by proletarian.job/enqueue! queue TEXT NOT NULL, -- queue name job_type TEXT NOT NULL, -- job type payload TEXT NOT NULL, -- Transit-encoded job data attempts INTEGER NOT NULL, -- Number of attempts. Starts at 0. Increments when the job is processed. enqueued_at TIMESTAMP NOT NULL, -- When the job was enqueued (never changes) process_at TIMESTAMP NOT NULL, -- When the job should be run (updates every retry) INDEX (queue(256), process_at));CREATE TABLE IF NOT EXISTS proletarian.archived_job ( job_id BINARY(16) PRIMARY KEY, -- Copied from job record. queue TEXT NOT NULL, -- Copied from job record. job_type TEXT NOT NULL, -- Copied from job record. payload TEXT NOT NULL, -- Copied from job record. attempts INTEGER NOT NULL, -- Copied from job record. enqueued_at TIMESTAMP NOT NULL, -- Copied from job record. process_at TIMESTAMP NOT NULL, -- Copied from job record (data for the last run only) status TEXT NOT NULL, -- success / failure finished_at TIMESTAMP NOT NULL -- When the job was finished (success or failure));
Note that job IDs are stored as BINARY(16) in MySQL. When using MySQL you must pass the :proletarian/job-id-strategy option to both worker/create-queue-worker and job/enqueue!:
(require '[proletarian.job :as job])(require '[proletarian.job-id-strategies :as job-id-strategies])(require '[proletarian.worker :as worker]);; When creating the worker:(worker/create-queue-worker ds handler-fn {:proletarian/job-id-strategy (job-id-strategies/->mysql-uuid-strategy)});; When enqueueing a job:(job/enqueue! conn ::my-job-type payload {:proletarian/job-id-strategy (job-id-strategies/->mysql-uuid-strategy)})
Use a migration library to manage these SQL files as part of your normal
schema lifecycle. Flyway and
Migratus are both popular choices in
the Clojure ecosystem. Copy the contents of the appropriate SQL block above
into a new migration file and commit it alongside your application code.
If you need to use different schema or table names — for example, because you have multiple applications sharing a single database — you can override the defaults everywhere Proletarian accepts options:
;; Use custom table names when creating the worker:(worker/create-queue-worker ds handler-fn {:proletarian/job-table "my_schema.jobs" :proletarian/archived-job-table "my_schema.archived_jobs"});; Use the same custom table name when enqueueing:(job/enqueue! conn ::my-job-type payload {:proletarian/job-table "my_schema.jobs"})
The values of :proletarian/job-table and :proletarian/archived-job-table must match the actual table names (including schema qualifier) in your database. Make sure your migration SQL uses the same names.
