Python: Multithreading e Multiprocessamento [Guia Completo]

Se você já precisou processar milhares de arquivos, fazer centenas de requisições HTTP ou acelerar um algoritmo pesado, sabe que o Python tradicional (síncrono) pode ser frustrantemente lento. A solução? Concorrência e paralelismo. Neste guia completo, você vai dominar multithreading e multiprocessamento em Python, entendendo quando e como usar cada técnica para extrair o máximo de performance do seu código.

Vamos explorar desde os conceitos fundamentais — como threads, processos e o famigerado GIL — até implementações práticas com ThreadPoolExecutor, ProcessPoolExecutor, locks e queues. Ao final, você terá um arsenal completo para escrever Python paralelo de verdade.

Entendendo Threads e Processos

Antes de escrevermos código, precisamos entender o que são threads e processos — e por que a diferença entre eles é crucial em Python.

Um processo é uma instância de um programa em execução. Cada processo tem seu próprio espaço de memória, seus próprios recursos e é isolado dos demais. Pense em abrir o Chrome três vezes: cada janela é um processo separado. Processos são pesados de criar, mas totalmente independentes.

Uma thread (ou linha de execução) é a menor unidade de processamento dentro de um processo. Todas as threads de um mesmo processo compartilham o mesmo espaço de memória. Pense em abas dentro de uma mesma janela do Chrome: todas compartilham recursos, mas podem travar umas às outras se não forem bem gerenciadas.

Em Python, essa distinção é ainda mais importante por causa de uma peculiaridade da implementação padrão: o GIL.

O GIL (Global Interpreter Lock)

O Global Interpreter Lock é um mecanismo do CPython (a implementação padrão do Python) que permite que apenas uma thread execute bytecode Python por vez. Isso significa que, em CPU-bound tasks (tarefas pesadas de processamento), múltiplas threads não trazem ganho de performance — na verdade, podem até piorar o desempenho devido ao overhead de alternância entre threads.

Mas então por que usar threads em Python? A resposta está nas I/O-bound tasks: operações de entrada e saída como requisições HTTP, leitura de arquivos ou consultas a bancos de dados. Durante essas operações, a thread fica ociosa esperando a resposta, e o GIL é liberado — permitindo que outras threads executem. É aí que o multithreading brilha.

Para entender mais a fundo o funcionamento do GIL, recomendo a leitura do artigo What Is the Python Global Interpreter Lock (GIL)? no Real Python, que explica o tema de forma clara e aprofundada.

Multithreading com threading.Thread

O módulo threading fornece a API básica para criar e gerenciar threads em Python. Vamos começar com um exemplo simples:

import threading
import time

def tarefa(nome, segundos):
    print(f"Thread {nome}: iniciando")
    time.sleep(segundos)
    print(f"Thread {nome}: concluída após {segundos}s")

threads = []
for i in range(5):
    t = threading.Thread(target=tarefa, args=(f"T{i}", i + 1))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print("Todas as threads concluídas!")

Chamamos start() para iniciar cada thread e join() para aguardar que todas terminem antes de prosseguir. Sem o join(), o programa principal poderia encerrar antes das threads filhas completarem seu trabalho.

A documentação oficial do módulo threading traz todos os detalhes sobre locks, semáforos, eventos e outros mecanismos de sincronização disponíveis.

Subclasses de Thread

Uma abordagem mais organizada é criar subclasses de Thread:

import threading

class MeuDownload(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.resultado = None

    def run(self):
        # Simula download
        import time
        time.sleep(2)
        self.resultado = f"Dados de {self.url}"

downloads = [MeuDownload(f"https://api.exemplo.com/item/{i}") for i in range(10)]
for d in downloads:
    d.start()
for d in downloads:
    d.join()
    print(d.resultado)

Essa abordagem é especialmente útil quando você precisa encapsular estado e comportamento em objetos reutilizáveis.

ThreadPoolExecutor: A Maneira Moderna

Gerenciar threads manualmente funciona, mas é trabalhoso. O módulo concurrent.futures, introduzido no Python 3.2 e significativamente melhorado ao longo das versões, oferece uma API muito mais elegante através do ThreadPoolExecutor:

from concurrent.futures import ThreadPoolExecutor
import time

def baixar_arquivo(url):
    print(f"Baixando: {url}")
    time.sleep(2)
    return f"Conteúdo de {url}"

urls = [f"https://site.com/arquivo_{i}.zip" for i in range(20)]

with ThreadPoolExecutor(max_workers=5) as executor:
    resultados = list(executor.map(baixar_arquivo, urls))

print(f"Baixados {len(resultados)} arquivos com sucesso!")

O ThreadPoolExecutor gerencia automaticamente um pool de threads reutilizáveis. O parâmetro max_workers define quantas threads serão mantidas no pool. Com o gerenciador de contexto (with), as threads são finalizadas automaticamente ao sair do bloco.

Você também pode usar submit() para obter objetos Future, que permitem acompanhar o progresso de cada tarefa individualmente:

from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(baixar_arquivo, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            resultado = future.result()
            print(f"{url} concluído: {resultado[:30]}...")
        except Exception as e:
            print(f"{url} falhou: {e}")

Consulte a documentação oficial do concurrent.futures para explorar todos os recursos avançados, incluindo callbacks e timeouts.

Sincronização: Locks e Queues

Quando múltiplas threads compartilham dados, surgem condições de corrida (race conditions). Duas threads podem tentar modificar a mesma variável ao mesmo tempo, corrompendo o resultado. A solução clássica é o uso de locks (travar e destravar o acesso a recursos compartilhados):

import threading

contador = 0
lock = threading.Lock()

def incrementar():
    global contador
    for _ in range(100000):
        with lock:
            contador += 1

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

print(f"Contador final: {contador}")  # Sempre 1.000.000

Sem o lock, o resultado seria imprevisível — provavelmente menor que 1.000.000 — devido às condições de corrida. O gerenciador de contexto with lock garante que o lock seja sempre liberado, mesmo em caso de exceções.

Para comunicação entre threads, a fila (queue.Queue) é a ferramenta ideal. Ela já é thread-safe, eliminando a necessidade de locks manuais:

from queue import Queue
import threading
import time

def produtor(fila):
    for i in range(10):
        item = f"Item {i}"
        fila.put(item)
        print(f"Produzido: {item}")
        time.sleep(0.5)
    fila.put(None)  # Sinal de fim

def consumidor(fila):
    while True:
        item = fila.get()
        if item is None:
            break
        print(f"Consumido: {item}")
        time.sleep(1)
    print("Consumidor finalizado")

fila = Queue()
t_produtor = threading.Thread(target=produtor, args=(fila,))
t_consumidor = threading.Thread(target=consumidor, args=(fila,))

t_produtor.start()
t_consumidor.start()
t_produtor.join()
t_consumidor.join()

O padrão produtor-consumidor com filas é extremamente versátil e aparece em aplicações reais como processamento de logs, web scraping e ETL. A documentação oficial do módulo queue detalha os tipos de fila disponíveis: FIFO, LIFO e filas com prioridade.

Para se aprofundar em mais padrões de concorrência, veja também nosso guia completo sobre context managers em Python, que mostra como usar o padrão with para gerenciar recursos de forma segura.

Multiprocessamento com ProcessPoolExecutor

Para tarefas que exigem muito processamento da CPU (CPU-bound) — como processamento de imagens, cálculos científicos ou machine learning — o multithreading não ajuda por causa do GIL. A solução é o multiprocessamento, que cria processos separados, cada um com seu próprio interpretador Python e seu próprio GIL.

O ProcessPoolExecutor do módulo concurrent.futures oferece a mesma API elegante do ThreadPoolExecutor, mas usando processos em vez de threads:

from concurrent.futures import ProcessPoolExecutor
import math

def calcular_primos(limite):
    primos = []
    for num in range(2, limite):
        if all(num % i != 0 for i in range(2, int(math.sqrt(num)) + 1)):
            primos.append(num)
    return len(primos)

intervalos = [10000, 20000, 30000, 40000, 50000, 60000]

with ProcessPoolExecutor(max_workers=4) as executor:
    resultados = list(executor.map(calcular_primos, intervalos))

for intervalo, total in zip(intervalos, resultados):
    print(f"Primos até {intervalo}: {total}")

A diferença de performance em tarefas CPU-bound é dramática. Enquanto threads seriam limitadas pelo GIL e executariam o trabalho praticamente de forma sequencial, processos distribuem o trabalho entre núcleos da CPU, alcançando ganhos lineares de desempenho.

É importante notar que ProcessPoolExecutor tem algumas limitações: os argumentos e retornos das funções precisam ser picklable (serializáveis), e o overhead de criação de processos é maior que o de threads. Para tarefas muito pequenas, o custo de criar o processo pode superar o benefício.

A documentação oficial do módulo multiprocessing oferece recursos mais avançados como filas entre processos, memória compartilhada e conexões pipes para cenários que exigem maior controle.

Outro recurso que pode complementar seu conhecimento é o guia sobre args e kwargs em Python, fundamental para criar funções flexíveis que funcionam bem com os executors.

Multithreading vs Multiprocessamento: Quando Usar Cada Um

Escolher entre threads e processos é uma das decisões mais importantes na programação concorrente em Python. Aqui vai um resumo prático:

Use Multithreading quando:

• A tarefa é I/O-bound (requisições HTTP, leitura de arquivos, consultas a banco de dados)
• Você precisa compartilhar estado entre tarefas com frequência
• O número de tarefas simultâneas é muito grande (milhares)
• O overhead de criação precisa ser mínimo

Use Multiprocessamento quando:

• A tarefa é CPU-bound (processamento de imagens, cálculos matemáticos)
• Você precisa aproveitar múltiplos núcleos da CPU
• A transferência de dados entre tarefas é pequena
• O isolamento entre processos é desejável (segurança, estabilidade)

O artigo Difference Between Multithreading vs Multiprocessing in Python no GeeksforGeeks oferece uma tabela comparativa detalhada que pode ajudar na decisão.

Exemplo Prático: Web Scraping Concorrente

Vamos juntar tudo em um exemplo realista: um web scraper que baixa múltiplas páginas simultaneamente:

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import time

def scrape_url(url):
    try:
        resp = requests.get(url, timeout=10)
        return url, resp.status_code, len(resp.text)
    except Exception as e:
        return url, None, str(e)

urls = [
    "https://python.org",
    "https://docs.python.org/3/",
    "https://pypi.org",
    "https://realpython.com",
    "https://github.com/python",
    "https://stackoverflow.com/questions/tagged/python",
    "https://www.geeksforgeeks.org/python-programming-language/",
    "https://pandas.pydata.org",
]

print("Iniciando scraping síncrono...")
inicio = time.time()
for url in urls:
    resultado = scrape_url(url)
    print(f"{resultado[0]}: {resultado[1]}")
print(f"Tempo síncrono: {time.time() - inicio:.2f}s")

print("\nIniciando scraping concorrente...")
inicio = time.time()
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(scrape_url, url): url for url in urls}
    for future in as_completed(futures):
        url, status, dados = future.result()
        print(f"{url}: {status}")
print(f"Tempo concorrente: {time.time() - inicio:.2f}s")

A diferença de tempo entre a versão síncrona e a concorrente é impressionante. Em testes reais, o scraping com ThreadPoolExecutor costuma ser de 3 a 8 vezes mais rápido, dependendo do número de URLs e da latência da rede.

O guia de concorrência em Python do Real Python expande esses exemplos com benchmarks e comparações detalhadas entre threading, multiprocessing e asyncio.

Boas Práticas e Armadilhas Comuns

Concorrência é poderosa, mas também traz desafios. Aqui estão as armadilhas mais comuns e como evitá-las:

1. Compartilhamento excessivo de estado: Quanto menos estado compartilhado entre threads/processos, melhor. Prefira passar dados como argumentos e retornar resultados, em vez de modificar variáveis globais.

2. Esquecer o join(): Sempre aguarde suas threads/processos terminarem com join() ou usando gerenciadores de contexto.

3. Deadlocks: Quando duas threads esperam uma pela outra liberar um lock. Use with lock e evite adquirir múltiplos locks ao mesmo tempo.

4. Excesso de workers: Criar threads ou processos demais degrada a performance. O número ideal de workers depende do tipo de tarefa: para I/O-bound, o dobro de threads do número de CPUs costuma funcionar bem; para CPU-bound, use o número de núcleos da sua máquina.

5. Ignorar exceções em threads: Exceções dentro de threads não propagam automaticamente para a thread principal. Sempre use future.result() ou try/except dentro da função alvo.

Para um estudo mais aprofundado sobre multithreading em Python, o tutorial Multithreading in Python do GeeksforGeeks cobre desde o básico até tópicos avançados como semáforos e barreiras de sincronização.

Considerações de Performance

Aqui vai uma tabela prática de referência para escolher a ferramenta certa:

Conclusão

Dominar multithreading e multiprocessamento é essencial para qualquer desenvolvedor Python que deseja escrever aplicações performáticas e escaláveis. Neste guia, você aprendeu:

• A diferença fundamental entre threads (mesmo espaço de memória) e processos (espaço isolado)
• Como o GIL impacta cada abordagem
• A usar ThreadPoolExecutor para I/O-bound tasks
• A usar ProcessPoolExecutor para CPU-bound tasks
• Técnicas de sincronização com locks e queues
• Padrões práticos como produtor-consumidor
• Exemplos reais de web scraping concorrente

O segredo do código concorrente bem-sucedido está em escolher a ferramenta certa para cada tipo de tarefa. Comece com o ThreadPoolExecutor para operações de I/O e evolua para o ProcessPoolExecutor quando precisar processar dados pesados. E lembre-se: meça sempre o desempenho antes e depois — a otimização prematura é a raiz de todos os males!

Gostou do guia? Explore também outros tutoriais no blog Universo Python para continuar evoluindo suas habilidades em programação concorrente e outros tópicos avançados da linguagem.