Python Threading: Threads, Locks e Guia de Concorrência

O módulo threading do Python permite executar múltiplas tarefas no mesmo processo ao mesmo tempo. Cada tarefa é executada em sua própria thread — uma unidade leve de execução que compartilha o espaço de memória do processo. Threading é a ferramenta certa quando o seu programa passa a maior parte do tempo aguardando (lendo um arquivo, fazendo uma requisição HTTP, consultando um banco de dados) e você quer realizar trabalho útil durante essa espera em vez de ficar bloqueado.

Este capítulo aborda:

Criando e iniciando threads com threading.Thread
Aguardando o término das threads com join
Threads daemon e tarefas em segundo plano
Prevenindo corridas de dados com Lock e with
Coordenando threads com Event e Semaphore
Comunicação thread-safe usando queue.Queue
O ThreadPoolExecutor para pools de threads gerenciados
O Global Interpreter Lock (GIL) e por que threads não aceleram código CPU-bound
Quando escolher threading vs. asyncio

Criando e iniciando uma thread

Importe threading e crie um objeto Thread, passando a função a executar como target. Chame .start() para lançar a thread:

import threading
import time

def greet(name):
    time.sleep(0.5)         # simulate some work
    print(f'Hello, {name}!')

t = threading.Thread(target=greet, args=('Alice',))
t.start()
print('Thread started — main continues running')
t.join()                    # wait for the thread to finish
print('Thread finished')
# Thread started — main continues running
# Hello, Alice!
# Thread finished

Pontos principais:

args é uma tupla de argumentos posicionais passados ao target. Use kwargs para argumentos nomeados.
Sem .join(), a thread principal pode encerrar antes que a thread criada termine.
.start() retorna imediatamente; a nova thread é executada de forma concorrente.

Passando argumentos nomeados

import threading

def connect(host, port=80):
    print(f'Connecting to {host}:{port}')

t = threading.Thread(target=connect, kwargs={'host': 'example.com', 'port': 443})
t.start()
t.join()
# Connecting to example.com:443

Executando múltiplas threads ao mesmo tempo

O verdadeiro benefício do threading é executar várias tarefas em paralelo. Crie todas as threads primeiro, depois aguarde todas elas:

import threading
import time

def download(url):
    time.sleep(1)           # simulate a 1-second network request
    print(f'Downloaded: {url}')

urls = [
    'https://example.com/data1',
    'https://example.com/data2',
    'https://example.com/data3',
]

start = time.perf_counter()

threads = [threading.Thread(target=download, args=(url,)) for url in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()

elapsed = time.perf_counter() - start
print(f'All downloads finished in {elapsed:.1f}s')
# Downloaded: https://example.com/data1
# Downloaded: https://example.com/data2
# Downloaded: https://example.com/data3
# All downloads finished in 1.0s

Sem threads, isso levaria 3 segundos (sequencial). Com três threads, leva cerca de 1 segundo porque as esperas se sobrepõem.

Criando subclasse de Thread

Para lógica mais complexa, crie uma subclasse de threading.Thread e substitua run(). Armazene os resultados como atributos de instância para que o código chamador possa lê-los após join():

import threading
import time

class DownloadThread(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = None

    def run(self):
        time.sleep(0.5)     # simulate download
        self.result = f'Data from {self.url}'

threads = [DownloadThread(f'https://example.com/page{i}') for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

for t in threads:
    print(t.result)
# Data from https://example.com/page0
# Data from https://example.com/page1
# Data from https://example.com/page2

Threads daemon

Uma thread daemon é uma thread em segundo plano que o interpretador encerra automaticamente quando todas as threads não-daemon foram finalizadas. Marque uma thread como daemon passando daemon=True (ou definindo t.daemon = True antes de .start()):

import threading
import time

def heartbeat():
    while True:
        print('♥ still running')
        time.sleep(1)

t = threading.Thread(target=heartbeat, daemon=True)
t.start()

time.sleep(2.5)
print('Main thread exiting — daemon will be killed')
# ♥ still running
# ♥ still running
# Main thread exiting — daemon will be killed

Use threads daemon para tarefas de monitoramento ou registro em segundo plano que não devem impedir o programa de encerrar. Nunca as use para tarefas que precisam ser concluídas de forma limpa (escrita em arquivo, commits em banco de dados) — elas são encerradas sem nenhuma limpeza.

Nomes de threads e introspecção

Cada thread tem um nome. Você pode defini-lo explicitamente ou deixar o Python atribuir um automaticamente. Use threading.current_thread() para inspecionar a thread em execução e threading.active_count() para contar as threads ativas:

import threading

def worker():
    t = threading.current_thread()
    print(f'Running in thread: {t.name}')

t = threading.Thread(target=worker, name='WorkerThread-1')
t.start()
t.join()

print(f'Active threads: {threading.active_count()}')
# Running in thread: WorkerThread-1
# Active threads: 1

Sincronização: prevenindo corridas de dados

Threads compartilham a memória do processo. Quando duas threads leem e escrevem a mesma variável simultaneamente, o resultado é uma corrida de dados — comportamento não determinístico que é difícil de reproduzir ou depurar.

O exemplo a seguir sem lock produz uma contagem final imprevisível porque os incrementos de threads diferentes podem se sobrepor:

import threading

counter = 0

def unsafe_increment():
    global counter
    for _ in range(100_000):
        counter += 1         # read-modify-write: not atomic!

threads = [threading.Thread(target=unsafe_increment) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

# counter is somewhere between 100000 and 500000 — unpredictable
print('Final counter:', counter)

Lock

Um threading.Lock garante que apenas uma thread execute a seção protegida por vez. Use-o como gerenciador de contexto com with para que o lock seja sempre liberado, mesmo que uma exceção seja lançada:

import threading

counter = 0
lock = threading.Lock()

def safe_increment():
    global counter
    for _ in range(100_000):
        with lock:              # acquire before read-modify-write
            counter += 1        # now only one thread at a time can run this

threads = [threading.Thread(target=safe_increment) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print('Final counter:', counter)   # always 500000

RLock (lock re-entrante)

Se uma thread precisa adquirir o mesmo lock duas vezes (por exemplo, um método chama outro método que também adquire o lock), use threading.RLock. Ele permite que a mesma thread readquira o lock sem causar deadlock:

import threading

lock = threading.RLock()

def outer():
    with lock:
        print('Outer acquired')
        inner()             # inner also acquires the same lock

def inner():
    with lock:              # works because RLock counts acquisitions
        print('Inner acquired')

t = threading.Thread(target=outer)
t.start()
t.join()
# Outer acquired
# Inner acquired

Coordenando threads: Event e Semaphore

Event

threading.Event é um sinal simples. Uma thread chama .set() para sinalizar; outras threads chamam .wait() para bloquear até que o sinal chegue:

import threading
import time

ready = threading.Event()

def worker():
    print('Worker: waiting for signal...')
    ready.wait()            # blocks here until ready.set() is called
    print('Worker: signal received, starting work')

t = threading.Thread(target=worker)
t.start()

time.sleep(0.5)
print('Main: sending signal')
ready.set()
t.join()
# Worker: waiting for signal...
# Main: sending signal
# Worker: signal received, starting work

Use um Event para coordenar a ordem de inicialização — por exemplo, para atrasar threads de trabalho até que uma conexão de banco de dados seja estabelecida.

Semaphore

Um threading.Semaphore limita o número de threads que podem estar em uma seção simultaneamente. Isso é útil para limitar o acesso a um recurso compartilhado, como um pool de conexões:

import threading
import time

# Allow at most 2 threads to enter the critical section at once
semaphore = threading.Semaphore(2)

def use_connection(name):
    with semaphore:
        print(f'{name}: using connection')
        time.sleep(0.5)
        print(f'{name}: releasing connection')

threads = [threading.Thread(target=use_connection, args=(f'T{i}',)) for i in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
# T0: using connection
# T1: using connection          <- only 2 at a time
# T0: releasing connection
# T2: using connection
# T1: releasing connection
# T3: using connection
# T2: releasing connection
# T3: releasing connection

Dados locais de thread

threading.local() cria um objeto que armazena valores separados por thread. Isso é útil para caches por thread ou cursores de banco de dados:

import threading

local_data = threading.local()

def set_user(name):
    local_data.user = name       # each thread writes its own copy
    print(f'{threading.current_thread().name}: user = {local_data.user}')

threads = [
    threading.Thread(target=set_user, args=(f'user{i}',), name=f'Thread-{i}')
    for i in range(3)
]
for t in threads:
    t.start()
for t in threads:
    t.join()
# Thread-0: user = user0
# Thread-1: user = user1
# Thread-2: user = user2

Ler local_data.user em uma thread onde nunca foi definido lança AttributeError, assim como qualquer outro acesso a atributo.

Filas thread-safe

A classe queue.Queue (da biblioteca padrão do módulo queue, não de asyncio) é um FIFO thread-safe. Threads podem usar put e get para itens sem um lock — toda a sincronização é tratada internamente.

O padrão clássico é produtor-consumidor: uma ou mais threads produtoras geram trabalho, threads consumidoras o processam:

import threading
import queue
import time

q = queue.Queue(maxsize=5)

def producer():
    for i in range(1, 5):
        q.put(f'item-{i}')
        print(f'Produced item-{i}')
        time.sleep(0.05)

def consumer():
    while True:
        item = q.get()
        if item is None:        # sentinel: stop when None is received
            break
        print(f'Consumed {item}')
        q.task_done()

prod = threading.Thread(target=producer)
cons = threading.Thread(target=consumer)

cons.start()
prod.start()
prod.join()
q.put(None)     # signal consumer to stop
cons.join()
# Produced item-1
# Consumed item-1
# Produced item-2
# Consumed item-2
# Produced item-3
# Consumed item-3
# Produced item-4
# Consumed item-4

queue.Queue também oferece task_done() e join() para rastrear quando todos os itens enfileirados foram processados, além de queue.LifoQueue / queue.PriorityQueue para ordenações alternativas.

ThreadPoolExecutor: pools de threads gerenciados

Criar um novo objeto Thread para cada tarefa é ineficiente quando você tem muitas tarefas de curta duração. concurrent.futures.ThreadPoolExecutor gerencia um pool de threads de trabalho reutilizáveis e retorna objetos Future para cada tarefa enviada:

import concurrent.futures
import time

def fetch_url(url):
    time.sleep(0.5)     # simulate network I/O
    return f'Response from {url}'

urls = [
    'https://api.example.com/users',
    'https://api.example.com/posts',
    'https://api.example.com/comments',
]

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # submit all tasks and get Future objects
    futures = {executor.submit(fetch_url, url): url for url in urls}

    for future in concurrent.futures.as_completed(futures):
        url = futures[future]
        print(future.result())

# Response from https://api.example.com/users    (order may vary)
# Response from https://api.example.com/comments
# Response from https://api.example.com/posts

executor.map(fn, iterable) é uma forma mais curta quando você não precisa de objetos Future individuais:

import concurrent.futures
import time

def square(n):
    time.sleep(0.01)
    return n * n

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(square, range(10)))

print(results)
# [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

executor.map preserva a ordem de entrada na saída, ao contrário de as_completed, que produz resultados na ordem de conclusão.

O Global Interpreter Lock (GIL)

O CPython (o interpretador Python padrão) possui um Global Interpreter Lock — um mutex que permite que apenas uma thread execute bytecode Python por vez. Isso significa que threads no CPython não podem executar código Python em verdadeiro paralelismo em múltiplos núcleos de CPU.

A implicação prática:

Tarefas I/O-bound: threads genuinamente aceleram o programa. Enquanto uma thread aguarda uma resposta de rede, o GIL é liberado e outra thread é executada. Todos os exemplos acima demonstram esse comportamento.
Tarefas CPU-bound: threads não aceleram as coisas e podem até ser ligeiramente mais lentas devido à sobrecarga de troca de contexto.

import threading
import time

def cpu_bound(n):
    total = 0
    for i in range(n):
        total += i
    return total

# Sequential
start = time.perf_counter()
cpu_bound(5_000_000)
cpu_bound(5_000_000)
single = time.perf_counter() - start

# Two threads — GIL prevents true parallelism
start = time.perf_counter()
t1 = threading.Thread(target=cpu_bound, args=(5_000_000,))
t2 = threading.Thread(target=cpu_bound, args=(5_000_000,))
t1.start(); t2.start()
t1.join(); t2.join()
threaded = time.perf_counter() - start

print(f'Single-threaded: {single:.2f}s')
print(f'Two threads:     {threaded:.2f}s')
# Two threads are NOT faster (similar elapsed time)

Para verdadeiro paralelismo de CPU em Python, use multiprocessing ou concurrent.futures.ProcessPoolExecutor — cada processo tem seu próprio GIL.

Threading vs. asyncio

Tanto threading quanto asyncio tornam programas I/O-bound mais rápidos, mas funcionam de forma diferente:

	`threading`	`asyncio`
Modelo de concorrência	Preemptivo — o SO troca as threads	Cooperativo — corrotinas cedem no `await`
Ideal para	Bibliotecas de terceiros bloqueantes	Bibliotecas com suporte async (`aiohttp`, `asyncpg`)
Estado compartilhado	Requer locks explícitos	Seguro dentro de um único loop de eventos
Sobrecarga	Uma thread de SO por tarefa	Muito baixa — milhares de corrotinas em uma thread
Curva de aprendizado	Familiar (código no estilo síncrono)	Requer `async/await` em todo o código

Regra geral: se você está usando uma biblioteca que tem uma versão compatível com async (por exemplo, aiohttp em vez de requests), opte por asyncio. Se você está preso com bibliotecas síncronas bloqueantes, use threading. Para trabalho CPU-bound, use multiprocessing.

Armadilhas comuns

Iniciar uma thread duas vezes. Chamar .start() no mesmo objeto Thread mais de uma vez lança RuntimeError. Crie uma nova instância de Thread para cada execução.

Esquecer de fazer join. Uma thread que não recebe join pode ainda estar em execução quando o programa encerra. Sempre faça join nas threads cuja conclusão importa, ou torne-as daemons se realmente forem do tipo "dispare e esqueça".

Manter um lock por muito tempo. Bloquear um grande bloco de código derrota o propósito da concorrência. Mantenha as seções bloqueadas tão curtas quanto possível — proteja apenas a operação real de leitura-modificação-escrita.

Deadlock. Um deadlock ocorre quando duas threads, cada uma segurando um lock que a outra está aguardando. Previna-o sempre adquirindo múltiplos locks na mesma ordem em todas as threads.

import threading

lock_a = threading.Lock()
lock_b = threading.Lock()

# DEADLOCK: Thread 1 holds lock_a, waits for lock_b
#           Thread 2 holds lock_b, waits for lock_a

# FIX: always acquire locks in the same order (lock_a then lock_b) in every thread

Modificar uma lista enquanto itera em outra thread. Envolva todo o acesso (leitura e escrita) a coleções compartilhadas com um lock para evitar RuntimeError: list changed size during iteration.

Resumo de referência rápida

Ferramenta	Finalidade
`threading.Thread(target=fn, args=(...))`	Criar uma nova thread
`t.start()`	Lançar a thread
`t.join()`	Aguardar o término da thread
`t.daemon = True`	Marcar como thread em segundo plano (encerrada ao sair)
`threading.Lock()`	Exclusão mútua — apenas uma thread por vez
`threading.RLock()`	Lock re-entrante — mesma thread pode adquirir múltiplas vezes
`threading.Event()`	Sinal único entre threads
`threading.Semaphore(n)`	Limitar a n threads concorrentes em uma seção
`threading.local()`	Armazenamento por thread
`queue.Queue`	FIFO thread-safe para padrões produtor-consumidor
`ThreadPoolExecutor(max_workers=n)`	Pool gerenciado de threads de trabalho reutilizáveis

Prática

Which of the following tasks would benefit most from Python threading?

Computing the first 1 million digits of pi using a pure-Python loop.Downloading 100 files from the internet at the same time.Sorting a large in-memory list using a CPU-intensive algorithm.Training a neural network with matrix multiplications.