Create and Manage Python Threads

Creating a Python thread is the simplest part of concurrent programming. The threading.Thread constructor takes a callable and arguments, and calling start() launches the thread. The complexity comes from what happens next: waiting for threads to finish, collecting their results, handling exceptions that occur in worker threads, and managing the lifecycle of many threads without creating more than your system can handle.

This article shows you how to create threads correctly from the first line of code and introduces ThreadPoolExecutor as the higher-level abstraction you should reach for in most real programs. If you have read about how Python threading works, you already understand the thread lifecycle and GIL interaction; this article focuses on the practical patterns you write every day.

The threading module gives you two levels of API. The low-level API centers on the Thread class, which you instantiate, configure, start, and join manually. The high-level API lives in concurrent.futures.ThreadPoolExecutor, which manages a pool of worker threads for you.

Most programs should use ThreadPoolExecutor because it handles thread lifecycle, result collection, and exception propagation automatically. Use raw Thread objects when you need a long-lived background thread, like a daemon monitor, or when you need explicit control over thread scheduling that a pool cannot provide.

Creating threads with the Thread class

The simplest way to create a thread is to instantiate threading.Thread with a target function and call start(). The target function runs in the new thread, and the main thread continues executing immediately after start() returns.

python

import threading
import time
 
def download_file(filename, duration):
    print(f"Downloading {filename}...")
    time.sleep(duration)
    print(f"{filename} downloaded.")
 
t1 = threading.Thread(target=download_file, args=("report.pdf", 2))
t2 = threading.Thread(target=download_file, args=("image.png", 1))
 
t1.start()
t2.start()
 
print("Downloads started. Main thread continues.")

Both downloads start immediately and the main thread's message prints before either download finishes, since start() returns right away instead of waiting for the target function to complete:

text

Downloading report.pdf...
Downloading image.png...
Downloads started. Main thread continues.
image.png downloaded.
report.pdf downloaded.

The two downloads run concurrently. The main thread prints its message and reaches the end of the script, but the program does not exit until both worker threads finish because both are non-daemon threads. Threads created with the default daemon=False keep the process alive.

If you do not need to wait for the threads to complete, set daemon=True and the program exits as soon as the main thread finishes. But in most cases you need the results of the worker threads, which means you must wait for them with join().

Waiting for threads with join()

join() blocks the calling thread until the target thread finishes. It is the mechanism for synchronizing the main thread with its workers. Without join(), you cannot safely access results that worker threads produce because you have no guarantee the work is done.

python

import threading
import time
 
results = {}
 
def fetch_data(source, duration):
    time.sleep(duration)
    results[source] = f"Data from {source}"
 
threads = [
    threading.Thread(target=fetch_data, args=("API", 2)),
    threading.Thread(target=fetch_data, args=("DB", 1)),
]
 
start = time.perf_counter()
 
for t in threads:
    t.start()
 
for t in threads:
    t.join()
 
print(f"Results: {results}")
print(f"Total: {time.perf_counter() - start:.2f}s")

Both join() calls block until their thread finishes, so by the time the program prints results, both dictionary entries are guaranteed to be filled in and the total elapsed time matches the longer of the two sleeps:

text

Results: {'API': 'Data from API', 'DB': 'Data from DB'}
Total: 2.00s

join() accepts an optional timeout in seconds. If the thread does not finish within the timeout, join() returns anyway, but the thread continues running. You can check t.is_alive() after a timed join to determine whether the thread is still working.

python

t = threading.Thread(target=download_file, args=("large.zip", 60))
t.start()
t.join(timeout=10)
 
if t.is_alive():
    print("Download is still in progress after 10 seconds.")
else:
    print("Download finished within timeout.")

This pattern is useful for implementing timeouts, progress monitoring, and graceful degradation when a worker thread takes longer than expected. The thread is not forcefully stopped; it continues running. Python does not provide a safe way to kill a thread from the outside, which is by design.

Forcing a thread to stop can leave locks held and resources in an inconsistent state.

Subclassing Thread

You can also create threads by subclassing Thread and overriding the run() method. This pattern is useful when the thread needs to maintain its own state or when the thread's behaviour is complex enough to warrant a class.

python

import threading
 
class DataProcessor(threading.Thread):
    def __init__(self, data, name=None):
        super().__init__(name=name)
        self.data = data
        self.result = None
 
    def run(self):
        total = sum(self.data)
        self.result = total * 2
 
processor = DataProcessor([1, 2, 3, 4, 5], name="Processor-1")
processor.start()
processor.join()
print(f"Result: {processor.result}")

Because run() stores its result on self.result, the value is still available on the processor object after join() returns, without needing a separate shared variable or queue:

text

Result: 30

The subclass pattern makes result collection straightforward because you can store the result as an instance attribute and read it after join(). The tradeoff is that subclassing couples the threading mechanism to the business logic. For most use cases, passing a target function to Thread and collecting results through a concurrent.futures.Future or a queue.Queue is more composable.

Using ThreadPoolExecutor for task-based concurrency

ThreadPoolExecutor from concurrent.futures is the recommended API for most threaded Python programs. You create an executor with a fixed number of worker threads, submit callables to it, and receive Future objects that represent the eventual results. The executor handles thread creation, reuse, and cleanup automatically.

python

from concurrent.futures import ThreadPoolExecutor
import time
 
def fetch_url(url):
    time.sleep(1)
    return f"Content of {url}"
 
urls = ["https://a.example.com", "https://b.example.com", "https://c.example.com"]
 
start = time.perf_counter()
 
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(fetch_url, url) for url in urls]
    results = [f.result() for f in futures]
 
print(results)
print(f"Total: {time.perf_counter() - start:.2f}s")

The three requests run concurrently on the pool's three worker threads, so the total time is close to one second instead of three, even though [f.result() for f in futures] waits for every future in order:

text

['Content of https://a.example.com', 'Content of https://b.example.com', 'Content of https://c.example.com']
Total: 1.01s

The with statement ensures the executor shuts down cleanly, waiting for all submitted tasks to complete before exiting the block. executor.submit() schedules a callable and returns immediately with a Future. Calling f.result() blocks until that specific future completes and returns its value.

If the callable raised an exception, result() re-raises it in the calling thread.

The executor pattern decouples task submission from execution. You submit tasks as fast as you can produce them, and the pool's worker threads pick them up at their own pace. This is more efficient than creating a new thread for each task and joining them individually.

Using executor.map() for simpler iteration

When you have a collection of inputs and one function to apply to each, executor.map() is simpler than executor.submit(). It applies the function to each input and returns results in the same order as the inputs.

python

from concurrent.futures import ThreadPoolExecutor
 
def square(n):
    return n * n
 
with ThreadPoolExecutor(max_workers=4) as executor:
    results = executor.map(square, [1, 2, 3, 4, 5])
 
print(list(results))

executor.map() hands each number to a worker thread and collects the squared results back in the same order the inputs were given, which is why the output lines up one-to-one with the input list:

text

[1, 4, 9, 16, 25]

executor.map() returns an iterator that yields results as they become available but preserves input order. If an input item raises an exception, the iterator raises that exception when you reach that position in the results. This behaviour makes map() ideal for batch processing where the order of results matters and each input is independent.

For cases where you want results as soon as any task completes regardless of submission order, use concurrent.futures.as_completed():

python

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random
 
def process_item(item):
    time.sleep(random.uniform(0.5, 2.0))
    return f"Processed {item}"
 
items = ["A", "B", "C", "D"]
 
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(process_item, item): item for item in items}
    for future in as_completed(futures):
        print(future.result())

Since each item sleeps for a different random duration, as_completed() yields whichever future finishes first, so the printed order reflects completion time rather than the order the items were submitted in:

text

Processed B
Processed D
Processed A
Processed C

as_completed() yields futures as they finish, which is useful when you want to start processing results immediately rather than waiting for all tasks to finish.

Handling exceptions in threads

Exceptions in raw Thread objects do not propagate to the main thread. The thread terminates, but the main thread has no built-in way to know that a failure occurred. There are three patterns for surfacing exceptions from worker threads.

The first pattern stores exceptions alongside results and checks them after join():

python

import threading
 
def risky_work(result_holder, error_holder):
    try:
        result = 1 / 0
        result_holder["value"] = result
    except Exception as e:
        error_holder["error"] = e
 
result = {}
error = {}
 
t = threading.Thread(target=risky_work, args=(result, error))
t.start()
t.join()
 
if error:
    print(f"Thread failed: {error['error']}")
else:
    print(f"Result: {result['value']}")

The second pattern uses concurrent.futures.ThreadPoolExecutor, which propagates exceptions through Future.result() automatically. Calling result() on a future whose task raised an exception re-raises that same exception in the thread that called result(), so you can catch it with an ordinary try/except block instead of building your own error-passing mechanism:

python

from concurrent.futures import ThreadPoolExecutor
 
def risky_work():
    return 1 / 0
 
with ThreadPoolExecutor() as executor:
    future = executor.submit(risky_work)
    try:
        result = future.result()
    except ZeroDivisionError as e:
        print(f"Task failed: {e}")

The third pattern sets threading.excepthook (available since Python 3.8) to define a global handler for uncaught thread exceptions. Python calls this hook automatically whenever a thread's target function raises without being caught, which makes it a good place to centralize logging for threads you do not want to wrap in try/except individually:

python

import threading
 
def custom_excepthook(args):
    print(f"Thread {args.thread.name} failed with {args.exc_type.__name__}: {args.exc_value}")
 
threading.excepthook = custom_excepthook
 
def failing_work():
    raise ValueError("Something went wrong")
 
t = threading.Thread(target=failing_work)
t.start()
t.join()

For most programs, ThreadPoolExecutor is the cleanest approach because it bakes exception propagation into the API. Raw thread exception handling is necessary only when you need long-lived threads that process many tasks over their lifetime.

Thread pools and the number of workers

Choosing max_workers for a ThreadPoolExecutor is a tuning decision. For I/O-bound work, the optimal number depends on how much time each task spends waiting versus computing. A task that spends 90% of its time waiting for network responses can support more concurrent workers than a task that spends 50% of its time computing.

The default max_workers in Python 3.8 and later is min(32, os.cpu_count() + 4). This default is reasonable for many I/O workloads but is intentionally conservative. If your tasks are extremely I/O-heavy, you can increase max_workers significantly.

If your tasks are compute-heavy, you should use ProcessPoolExecutor instead.

python

from concurrent.futures import ThreadPoolExecutor
import time
 
def io_task(duration):
    time.sleep(duration)
    return duration
 
durations = [0.1] * 20
 
for workers in [2, 5, 10, 20]:
    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=workers) as executor:
        list(executor.map(io_task, durations))
    elapsed = time.perf_counter() - start
    print(f"workers={workers:>2}: {elapsed:.2f}s")

The loop reruns the same 20 tasks with an increasing worker count each time, and the printed timings show how the total runtime drops as more workers can run the sleeps in parallel. Typical output for 20 tasks each sleeping 0.1 seconds looks like this:

text

workers= 2: 1.00s
workers= 5: 0.40s
workers=10: 0.20s
workers=20: 0.10s

With 20 workers, all 20 tasks run concurrently and finish in 0.1 seconds. With 2 workers, the pool processes 2 tasks at a time, taking 1 second total. The optimal value is the smallest number that keeps all CPU cores busy without wasting memory on idle threads.

Benchmark with your actual workload and data sizes to find the right number.

Managing thread lifecycles in long-running programs

Programs that run for hours or days need more careful thread management than scripts that finish in seconds. Thread pools should be created once and reused, not created and destroyed for every batch of work. Long-lived daemon threads need health monitoring to detect when they hang or crash silently.

For a long-running service, create a single ThreadPoolExecutor at startup and use it throughout the program's lifetime:

python

from concurrent.futures import ThreadPoolExecutor
import atexit
 
executor = ThreadPoolExecutor(max_workers=10, thread_name_prefix="Worker")
 
def shutdown():
    executor.shutdown(wait=True)
 
atexit.register(shutdown)

The atexit handler ensures the executor shuts down gracefully when the program exits, waiting for in-flight tasks to complete. Calling executor.shutdown(wait=True) is equivalent to the with statement's cleanup but can be called at any point in the program's lifecycle.

For long-lived individual threads that run background work in a loop, wrap the loop body in a try-except that logs failures and continues rather than letting the thread die silently:

python

import threading
import logging
import time
 
def background_monitor():
    while True:
        try:
            check_system_health()
        except Exception:
            logging.exception("Health check failed")
        time.sleep(30)
 
monitor = threading.Thread(
    target=background_monitor,
    daemon=True,
    name="HealthMonitor",
)
monitor.start()

This pattern prevents a single transient failure from killing the background thread. For more robust health monitoring, store the last successful check time and have a separate watchdog that alerts you if a background thread stops producing work.

Communicating between threads safely

When multiple threads share data, you need a safe communication mechanism. The simplest safe pattern is queue.Queue, which is thread-safe by design and does not require external locking.

python

import threading
import queue
import time
 
def producer(q):
    for i in range(5):
        q.put(i)
        time.sleep(0.1)
    q.put(None)
 
def consumer(q, name):
    while True:
        item = q.get()
        if item is None:
            q.put(None)
            break
        print(f"{name} got {item}")
        q.task_done()

The None value acts as a poison pill: when a consumer sees it, it puts None back for the other consumers to find and then exits its loop. This lets one producer signal an unknown number of consumers to stop without them needing to coordinate directly.

python

q = queue.Queue()
 
threads = [
    threading.Thread(target=producer, args=(q,)),
    threading.Thread(target=consumer, args=(q, "Consumer-1")),
    threading.Thread(target=consumer, args=(q, "Consumer-2")),
]
 
for t in threads:
    t.start()
for t in threads:
    t.join()

The Queue handles all synchronization internally. put() blocks if the queue is full (when a maximum size is set), and get() blocks if the queue is empty. This makes queues the preferred communication mechanism for producer-consumer patterns in threaded Python programs.

Later articles in this section cover thread communication with queues in more detail.