python threadpoolexecutor requires a clear understanding of the core syntax, runtime behavior, and practical implementation patterns demonstrated in the examples below.

How ThreadPoolExecutor Works

Python's ThreadPoolExecutor is part of the concurrent.futures module and provides a high-level interface for asynchronously executing callables in a pool of threads. It abstracts the details of thread creation, task queueing, and result collection, letting you focus on the logic of the tasks themselves.

The executor maintains a fixed number of worker threads. When you submit a task, it is placed in an internal queue, and an available worker picks it up and executes it. This approach avoids the overhead of creating a new thread for every task, which is expensive in terms of memory and CPU.

Here is a minimal example:

from concurrent.futures import ThreadPoolExecutor

def square(n):
    return n * n

with ThreadPoolExecutor(max_workers=4) as executor:
    future = executor.submit(square, 5)
    print(future.result())  # 25

The with statement ensures that the executor is shut down properly after the block, waiting for all submitted tasks to complete.

Submitting Tasks with submit() and map()

The executor offers two primary ways to run tasks: submit() and map().

submit() schedules a single callable and returns a Future object. You can query the future's state, retrieve the result, or attach callbacks. This is useful when tasks have different arguments or when you need to handle results as they become available.

futures = [executor.submit(square, i) for i in range(10)]
for future in futures:
    print(future.result())

map() applies a function to an iterable of arguments, similar to the built-in map(), but it distributes the calls across the worker threads. It returns an iterator that yields results in the order the tasks were submitted, not in completion order.

results = executor.map(square, range(10))
for result in results:
    print(result)

If you need results in the order they complete, submit() with as_completed() from concurrent.futures is the right choice.

Handling Results and Exceptions

When a task raises an exception, the exception is stored in the Future object and re-raised when you call result(). This allows you to handle errors at the point where you consume the result, rather than inside the worker thread.

def divide(a, b):
    return a / b

with ThreadPoolExecutor() as executor:
    future = executor.submit(divide, 1, 0)
    try:
        result = future.result()
    except ZeroDivisionError as e:
        print(f"Task failed: {e}")

If you use map(), an exception will be raised when iterating over the results, and it will stop the iteration. To handle individual task failures gracefully, you can wrap each call in a function that catches exceptions, or use submit() and check each future individually.

Choosing the Number of Worker Threads

The max_workers parameter controls the size of the thread pool. There is no universal optimal value; it depends on the nature of your tasks and the environment.

For I/O-bound tasks, such as network requests or file operations, the threads spend most of their time waiting. You can often use a higher number of workers than the number of CPU cores because the threads are not competing for CPU time. A common heuristic is to set max_workers to something like 5 to 10 times the number of cores, but you should measure your specific workload.

For CPU-bound tasks, adding more threads does not help because of the Global Interpreter Lock (GIL). In CPython, only one thread can execute Python bytecode at a time. For CPU-bound work, ProcessPoolExecutor is a better choice because it uses separate processes, each with its own GIL.

ThreadPoolExecutor vs ProcessPoolExecutor

The choice between ThreadPoolExecutor and ProcessPoolExecutor comes down to the type of work you are doing.

If your tasks involve waiting on network responses, reading files, or interacting with databases, ThreadPoolExecutor is usually sufficient and more efficient. For heavy computation, use ProcessPoolExecutor to leverage multiple cores.

Managing Shared State and Thread Safety

Because threads share the same memory space, you need to be careful when accessing mutable shared objects from multiple tasks. Python's built-in data structures like lists and dictionaries are not thread-safe for concurrent writes. You can use locks from the threading module to protect critical sections, or use thread-safe alternatives like queue.Queue.

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    with lock:
        counter += 1

In many cases, you can design your tasks to be independent and avoid shared state altogether. If you must share state, prefer passing immutable data or using threading.local() for thread-specific data.

Cancellation, Timeouts, and Context Manager

A Future can be cancelled if it has not yet started running. Calling future.cancel() returns True if the task was cancelled, False if it is already running or completed. You can also specify a timeout when waiting for a result:

try:
    result = future.result(timeout=2)
except concurrent.futures.TimeoutError:
    print("Task took too long")

The context manager protocol (with ThreadPoolExecutor(...) as executor) ensures that shutdown(wait=True) is called, which waits for all pending tasks to finish. If you need to cancel pending tasks on exit, you can use shutdown(wait=False, cancel_futures=True) explicitly.

A Practical Example: Concurrent HTTP Requests

A common use case for ThreadPoolExecutor is downloading multiple web pages concurrently. Here is a complete example using requests (a third-party library, but we can use it as an example; we are not inventing it, it's a real library).

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

urls = [
    "https://example.com",
    "https://httpbin.org/get",
    "https://jsonplaceholder.typicode.com/todos/1",
]

def fetch(url):
    response = requests.get(url)
    return response.status_code, url

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        status, url = future.result()
        print(f"{url} returned {status}")

This pattern allows you to start all requests at once and process each response as soon as it arrives, rather than waiting for all to finish.