ye_olde_threads.py

import queue
import threading
import time
import random

NUM_WORKERS = 20

class Worker(threading.Thread):
    """A worker thread."""
    def __init__(self, input, output=None):
        self._get_job = input.get
        if output:
            self._put_job = output.put
        super().__init__()

    def run(self):
        """Get a job and process it. Stop when there's no more jobs"""
        while True:
            job = self._get_job()
            if job is None:
                break
            self._process_job(job)

    def _process_job(self, job):
        """Do useful work here."""
        time.sleep(random.random())
        result = job + 1
        self._put_job(result)


class Recorder(Worker):
    def _process_job(self, job):
        """Override Worker's _process_job method. Just print our input"""
        print(job)


def main():
    job_queue = queue.Queue(0)
    results_queue = queue.Queue(0)

    # Create our pool of worker threads
    workers = []
    for x in range(NUM_WORKERS):
        worker = Worker(job_queue, results_queue)
        worker.start()
        workers.append(worker)

    # Create our single recording thread
    recorder = Recorder(results_queue)
    recorder.start()

    # Give the workers some numbers to crunch
    for x in range(NUM_WORKERS*2):
        job_queue.put(x)

    # Insert end of job markers
    for x in range(NUM_WORKERS):
        job_queue.put(None)

    # Wait for all workers to end
    for worker in workers:
        worker.join()

    # Tell recording thread it can stop
    results_queue.put(None)
    recorder.join()


if __name__ == '__main__':
    main()