← thecodex.expert · The Codex Family of Knowledge
Tier 4 · Advanced · Python Project

Background Task Queue

Build a task queue where producers add jobs and worker threads process them in the background. The pattern behind Celery and every job system.

🧠 Teaches how to think spoonfed, every age Last verified:

1 The Problem

We want a task queue: one part of the program adds jobs (send an email, resize an image), and background workers pick them up and process them — so the main program never waits. It teaches the producer-consumer pattern with a thread-safe queue, the backbone of every background-job system.

Where this shows up: Celery, job queues, background processing, sending emails without blocking a web request, video encoding, report generation. Offloading slow work to background workers keeps apps responsive.

2 How to Think About It

Think about producers and consumers, before any code:

The plan — in plain English
1. A shared queue holds pending jobs. → 2. Producers add jobs to the queue and move on immediately. → 3. Workers (background threads) take jobs from the queue and run them, one after another. → The queue is thread-safe, so producers and workers never corrupt it even running at once.

Producer creates a job

Put job on the queue

Queue holds pending jobs

Worker takes a job

Worker runs it

More workers

3 The Build — explained part by part

Here is a complete task queue with worker threads. Each part is explained below.

Pythonqueue_worker.py
import queue
import threading
import time

class TaskQueue:
    def __init__(self, num_workers=2):
        self.jobs = queue.Queue()      # thread-safe queue
        self.results = []
        self.lock = threading.Lock()
        for _ in range(num_workers):
            threading.Thread(target=self._worker, daemon=True).start()

    def _worker(self):
        while True:
            func, args = self.jobs.get()      # waits until a job is available
            result = func(*args)
            with self.lock:                   # protect the shared results list
                self.results.append(result)
            self.jobs.task_done()

    def add(self, func, *args):
        self.jobs.put((func, args))

    def wait(self):
        self.jobs.join()      # block until all jobs are done

def square(n):
    time.sleep(0.1)           # pretend it is slow work
    return n * n

if __name__ == "__main__":
    q = TaskQueue(num_workers=3)
    for i in range(5):
        q.add(square, i)
    q.wait()
    print("Results:", sorted(q.results))
What each part does — in plain words
queue.Queue() — a thread-safe queue from the standard library. Multiple threads can put and get from it safely without us managing locks for the queue itself.

_worker — each worker loops forever: self.jobs.get() waits until a job appears, runs it, stores the result, and marks it done. get() blocking means idle workers use no CPU.

daemon=True — worker threads die automatically when the main program exits, so we do not have to shut them down manually.

with self.lock: — the results list is shared, and appending from multiple threads at once could corrupt it. The lock ensures only one worker writes at a time.

add(func, *args) — producers call this to enqueue a job: the function to run and its arguments.

wait() / jobs.join() — block until every queued job has been processed. task_done() in the worker is what makes join() know when all are finished.
Common mistakes — and how to avoid them
✗ Appending to a shared list from many threads without a lock — results get corrupted.
✓ Protect shared state with a threading.Lock.
✗ Busy-waiting for jobs in a loop — wastes CPU.
✓ Use queue.get(), which blocks efficiently until a job arrives.
✗ Expecting results in submission order — workers finish unpredictably.
✓ Do not rely on order; sort or tag results if order matters.

4 Test & Prove Each Part

We test that all jobs get processed and produce the right results, using a real queue but a quick task.

All added jobs are processed
Results are correct
Multiple workers handle the load
Pythontest_queue.py
import queue
import threading


class TaskQueue:
    def __init__(self, num_workers=2):
        self.jobs = queue.Queue()
        self.results = []
        self.lock = threading.Lock()
        for _ in range(num_workers):
            threading.Thread(target=self._worker, daemon=True).start()

    def _worker(self):
        while True:
            func, args = self.jobs.get()
            result = func(*args)
            with self.lock:
                self.results.append(result)
            self.jobs.task_done()

    def add(self, func, *args):
        self.jobs.put((func, args))

    def wait(self):
        self.jobs.join()


def double(n):
    return n * 2


def test_all_processed():
    """All 10 jobs are processed."""
    q = TaskQueue(3)
    for i in range(10):
        q.add(double, i)
    q.wait()
    assert len(q.results) == 10


def test_correct_results():
    """Results are correct (order may vary)."""
    q = TaskQueue(2)
    for i in range(5):
        q.add(double, i)
    q.wait()
    assert sorted(q.results) == [0, 2, 4, 6, 8]


def test_single_worker():
    """Even one worker processes everything."""
    q = TaskQueue(1)
    q.add(double, 21)
    q.wait()
    assert q.results == [42]

Run with pytest -v. We sort results before checking because workers finish in unpredictable order — a key insight about concurrency: you control what gets done, not the exact order.

5 The Interface

addq.add(func, *args)enqueue a job (returns immediately)
waitq.wait()block until all jobs done
Flow
q.add(send_email, "alice@x.com")  # returns now
# worker sends it in the background

6 Run It & Automate It

Run it locally
python3 queue_worker.py
Watch 3 workers process 5 jobs in parallel — faster than doing them one by one.

Jenkins runs the tests automatically — each line explained below.

What you should see when it works
Terminala real run
Results: [0, 1, 4, 9, 16]
If it breaks — how to fix it
🚨 wait() hangs forever.
Every get() needs a matching task_done(), or join() never returns.
🚨 Results are missing or wrong.
You likely have a race on the shared list. Add the lock around the append.
GroovyJenkinsfile
// Jenkinsfile — automated tests on every change.
pipeline {
    agent any
    stages {
        stage('Get the code') { steps { checkout scm } }       // download the code
        stage('Install') { steps { sh 'python3 -m venv venv && . venv/bin/activate && pip install pytest' } }  // test tool
        stage('Test') { steps { sh '. venv/bin/activate && pytest -v' } }   // run every test
    }
    post {
        success { echo 'All tests passed.' }
        failure { echo 'A test failed — look above.' }
    }
}
🎯 Try this next — make it yours
  1. Priorities. Use a PriorityQueue so urgent jobs run first. (Teaches: priority queues.)
  2. Retries. Re-queue jobs that fail. (Teaches: error handling in workers.)
  3. Persistence. Save the queue so jobs survive a restart. (Teaches: durable queues.)
What you learned
You built a task queue with the producer-consumer pattern: a thread-safe queue, background workers, and a lock protecting shared results. This is how Celery and every background-job system works. Related: queue, Concurrency.