Concorrência: threads, processos e async/await

Todo programa que vimos até aqui executa uma instrução por vez — modelo sequencial. Mas o mundo real exige mais: baixar múltiplos arquivos simultaneamente, processar milhares de requisições por segundo, aproveitar todos os núcleos do processador. Concorrência e paralelismo são os mecanismos que tornam isso possível, e Python oferece três abordagens distintas para isso, cada uma adequada a um tipo de problema.

Conceitos Fundamentais

Antes do código, é essencial distinguir três conceitos que são frequentemente confundidos:

Concorrência — múltiplas tarefas progridem ao mesmo tempo, alternando execução. Não necessariamente simultâneas.

Paralelismo — múltiplas tarefas executam literalmente ao mesmo tempo, em núcleos distintos do processador.

Assincronismo — uma tarefa cede o controle enquanto aguarda algo externo (I/O), permitindo que outras tarefas avancem.

Tipo de problema          Melhor abordagem
──────────────────────────────────────────
I/O intensivo (rede, disco)  → asyncio ou threading
CPU intensivo (cálculo)      → multiprocessing
Misto                        → multiprocessing + asyncio

O GIL: Global Interpreter Lock

Python tem uma limitação importante: o GIL — um mutex que garante que apenas uma thread Python execute bytecode por vez. Isso significa que threads em Python não entregam paralelismo real para tarefas CPU-intensivas.

Threading em Python:
  Thread A: executa → pausa → executa → pausa
  Thread B:         executa →         executa

Multiprocessing em Python:
  Processo A: ████████████████  (núcleo 0)
  Processo B: ████████████████  (núcleo 1)

Para I/O — onde a thread fica bloqueada esperando resposta da rede ou disco — o GIL é liberado, e threads funcionam bem. Para cálculo puro, use multiprocessing.

O GIL, porém, deixou de ser absoluto. A 3.13 trouxe o primeiro build sem GIL, e a 3.14 passou a dar suporte oficial ao modo free-threading — mas, como vimos em Dominando o Python, ele é uma instalação à parte, e não o padrão. O executável costuma se chamar python3.14t, e nele sys._is_gil_enabled() devolve False. Medido na 3.14.7 com a função de primos da seção de multiprocessing, logo adiante: com quatro threads, o build padrão levou 1,86 s contra 1,78 s do laço sequencial — ganho nenhum —, enquanto o build sem GIL desceu de 2,49 s para 1,15 s. A mesma medição mostra o preço: rodando numa thread só, o código ficou cerca de 40% mais lento sem o GIL, e boa parte das extensões em C ainda não publicou versão compatível. Para cálculo pesado, hoje, processos continuam sendo a resposta padrão.

threading: Threads para I/O

import threading
import time
import requests


def baixar_url(url, resultados, indice):
    """Função executada em cada thread."""
    inicio = time.perf_counter()
    try:
        resposta = requests.get(url, timeout=10)
    except requests.RequestException as e:
        # exceção dentro da thread não chega ao join(): trate aqui
        resultados[indice] = {"url": url, "erro": str(e)}
        return
    resultados[indice] = {
        "url":    url,
        "status": resposta.status_code,
        "bytes":  len(resposta.content),
        "tempo":  time.perf_counter() - inicio
    }


urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
]

# Com threads — ~1 segundo (em sequência seriam ~4)
inicio = time.perf_counter()
resultados = [None] * len(urls)

threads = []
for i, url in enumerate(urls):
    t = threading.Thread(target=baixar_url, args=(url, resultados, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()   # aguarda todas terminarem

total = time.perf_counter() - inicio
print(f"Paralelo com threads: {total:.2f}s")
for r in resultados:
    if "erro" in r:
        print(f"  falhou — {r['url']}: {r['erro'][:60]}")
    else:
        print(f"  {r['status']} — {r['bytes']} bytes em {r['tempo']:.2f}s")

O try dentro de baixar_url não é enfeite. Se uma URL falha, a exceção morre dentro da thread: o traceback aparece no terminal, mas o join() retorna normalmente e resultados[i] continua None. Sem o tratamento, quem quebra é o laço de impressão, com TypeError: 'NoneType' object is not subscriptable — longe da causa. O ThreadPoolExecutor, a seguir, resolve isso de fábrica: future.result() relança a exceção na thread que pediu o resultado.

ThreadPoolExecutor: Interface de Alto Nível

A forma moderna e recomendada de usar threads:

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


def buscar_usuario(user_id):
    url      = f"https://jsonplaceholder.typicode.com/users/{user_id}"
    resposta = requests.get(url, timeout=10)
    dados    = resposta.json()
    return {"id": dados["id"], "nome": dados["name"], "email": dados["email"]}


inicio = time.perf_counter()

with ThreadPoolExecutor(max_workers=5) as executor:
    # submit — submete tarefas e retorna futures
    futures = {
        executor.submit(buscar_usuario, uid): uid
        for uid in range(1, 11)
    }

    # as_completed — processa conforme terminam (não necessariamente em ordem)
    for future in as_completed(futures):
        uid = futures[future]
        try:
            resultado = future.result()
            print(f"  [{resultado['id']:2}] {resultado['nome']:25} — {resultado['email']}")
        except Exception as e:
            print(f"  [{uid}] Erro: {e}")

print(f"\nTotal: {time.perf_counter() - inicio:.2f}s")

Lock: Sincronização entre Threads

Quando múltiplas threads acessam o mesmo recurso, use Lock para evitar condições de corrida:

import threading

class ContadorSeguro:
    def __init__(self):
        self._valor = 0
        self._lock  = threading.Lock()

    def incrementar(self):
        with self._lock:   # garante acesso exclusivo
            self._valor += 1

    def valor(self):
        return self._valor


contador = ContadorSeguro()

def incrementar_mil():
    for _ in range(1000):
        contador.incrementar()

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

print(f"Esperado: 10000 | Obtido: {contador.valor()}")  # 10000

Tire o Lock e rode de novo: no Python 3.14 padrão o resultado continua 10000 — e continuou exato em todas as execuções que medimos, mesmo com dez threads somando um milhão cada. Isso não prova que += é seguro; prova que a corrida é rara. O interpretador só troca de thread em pontos específicos, e um self._valor += 1 quase nunca é cortado ao meio. Basta haver uma espera entre a leitura e a gravação — um time.sleep(0), um acesso a disco, uma chamada de rede — para o mesmo contador terminar perto de 1000 de 10000. E no build sem GIL, dez threads somando cem mil cada chegaram a uns 210 mil de 1 milhão. O Lock é o que torna o código correto em qualquer interpretador; a ausência de erro num teste pequeno não diz nada.

multiprocessing: Paralelismo Real

Para tarefas CPU-intensivas, cada processo tem seu próprio interpretador Python — sem GIL:

import multiprocessing
import time
import math


def calcular_primos(limite):
    """Verifica quais números até limite são primos."""
    primos = []
    for n in range(2, limite):
        eh_primo = all(n % i != 0 for i in range(2, int(math.sqrt(n)) + 1))
        if eh_primo:
            primos.append(n)
    return len(primos)


def benchmark():
    limites = [100_000, 200_000, 300_000, 400_000]

    # Sequencial
    inicio = time.perf_counter()
    resultados_seq = [calcular_primos(l) for l in limites]
    tempo_seq = time.perf_counter() - inicio

    # Paralelo
    inicio = time.perf_counter()
    with multiprocessing.Pool(processes=4) as pool:
        resultados_par = pool.map(calcular_primos, limites)
    tempo_par = time.perf_counter() - inicio

    print(f"Sequencial:  {tempo_seq:.2f}s")
    print(f"Paralelo:    {tempo_par:.2f}s")
    print(f"Speedup:     {tempo_seq / tempo_par:.1f}x")


if __name__ == "__main__":   # obrigatório no Windows, no macOS e, desde a 3.14, no Linux
    benchmark()

Não espere 4× com quatro processos. Medido numa máquina de 16 núcleos, o speedup foi de 1,5×, e o motivo está na carga: as quatro tarefas levam 0,12 s, 0,31 s, 0,54 s e 0,80 s, e o pool.map só termina quando a mais longa termina. Com quatro tarefas iguais, de 250 mil cada, o mesmo código chega a 3,6×. Divida o trabalho em pedaços de tamanho parecido — ou em muitos pedaços pequenos — antes de culpar o multiprocessing. E o if __name__ == "__main__" deixou de ser coisa de Windows: desde a 3.14 o Linux inicia os processos com forkserver, cada filho reimporta o módulo, e sem essa guarda o script termina em RuntimeError. O mesmo código rodava sem ela na 3.12.

ProcessPoolExecutor: Interface de Alto Nível

from concurrent.futures import ProcessPoolExecutor, as_completed
import math


def fatorar(n):
    """Fatoração em números primos — CPU intensivo."""
    fatores = []
    d = 2
    while d * d <= n:
        while n % d == 0:
            fatores.append(d)
            n //= d
        d += 1
    if n > 1:
        fatores.append(n)
    return fatores


# Cada um é o produto de dois primos, e o menor fica perto de 10 milhões:
# é o tamanho do laço até achá-lo que dá trabalho de verdade a cada processo.
numeros = [
    9_999_900_370_006_237,
    9_982_350_693_275_171,
    9_999_929_069_999_503,
    9_999_929_930_007_383,
]

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = {executor.submit(fatorar, n): n for n in numeros}
        for future in as_completed(futures):
            n       = futures[future]
            fatores = future.result()
            print(f"  {n} = {' × '.join(map(str, fatores))}")

A escolha dos números importa. Com primos no lugar deles — 999.999.937 ou 1.000.000.007, por exemplo —, cada fatoração termina em 2 milissegundos, e o pool leva 0,078 s contra 0,006 s do laço sequencial: 13× mais lento, porque criar os processos custa mais que o trabalho. Com os produtos de dois primos acima, o laço sequencial leva 3,0 s e o pool, 1,2 s. Processo só compensa quando cada tarefa dura bem mais que o custo de criá-lo e de mandar os dados de ida e volta.

asyncio: Assincronismo Cooperativo

asyncio é ideal para I/O massivo — milhares de conexões simultâneas com um único thread:

import asyncio
import httpx
import time


async def buscar_post(client, post_id):
    resposta = await client.get(
        f"https://jsonplaceholder.typicode.com/posts/{post_id}"
    )
    dados = resposta.json()
    return {"id": dados["id"], "titulo": dados["title"][:40]}


async def buscar_todos(ids):
    async with httpx.AsyncClient(timeout=15.0) as client:
        tarefas    = [buscar_post(client, i) for i in ids]
        resultados = await asyncio.gather(*tarefas)
        return resultados


async def main():
    inicio = time.perf_counter()
    posts  = await buscar_todos(range(1, 21))
    total  = time.perf_counter() - inicio

    for post in posts[:5]:
        print(f"  [{post['id']:2}] {post['titulo']}")
    print(f"  ...")
    print(f"\n20 posts em {total:.2f}s")


asyncio.run(main())

asyncio: Conceitos Essenciais

import asyncio


# Corrotina — função async
async def tarefa(nome, duracao):
    print(f"[{nome}] iniciando...")
    await asyncio.sleep(duracao)   # cede o controle — não bloqueia
    print(f"[{nome}] concluída após {duracao}s")
    return f"resultado_{nome}"


# gather — executa as corrotinas concorrentemente, num único thread
async def demo_gather():
    inicio     = asyncio.get_running_loop().time()
    resultados = await asyncio.gather(
        tarefa("A", 1.0),
        tarefa("B", 1.5),
        tarefa("C", 0.5),
    )
    total = asyncio.get_running_loop().time() - inicio
    print(f"Todos concluídos em {total:.2f}s")   # ~1.5s, não 3.0s
    print(resultados)


# TaskGroup — Python 3.11+ — mais elegante
async def demo_taskgroup():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(tarefa("X", 1.0))
        t2 = tg.create_task(tarefa("Y", 0.5))

    print(t1.result())
    print(t2.result())


# timeout — cancela se demorar demais
async def com_timeout():
    try:
        async with asyncio.timeout(0.3):   # Python 3.11+
            await tarefa("lenta", 2.0)
    except TimeoutError:   # asyncio.TimeoutError é apelido dele desde a 3.11
        print("Operação cancelada por timeout.")


asyncio.run(demo_gather())
asyncio.run(demo_taskgroup())
asyncio.run(com_timeout())

Queue Assíncrona: Produtor e Consumidor

Padrão clássico para processar itens em pipeline:

import asyncio
import httpx


N_CONSUMIDORES = 3


async def produtor(fila: asyncio.Queue, urls: list):
    for url in urls:
        await fila.put(url)
        print(f"  [Produtor] enfileirou: {url}")
    # Um sinal de fim para CADA consumidor: se faltar um, ele espera para sempre
    for _ in range(N_CONSUMIDORES):
        await fila.put(None)


async def consumidor(nome, fila: asyncio.Queue, client: httpx.AsyncClient):
    while True:
        url = await fila.get()
        if url is None:
            print(f"  [{nome}] encerrando.")
            break
        try:
            r = await client.get(url, timeout=5.0)
            print(f"  [{nome}] {r.status_code} — {url}")
        except Exception as e:
            print(f"  [{nome}] Erro: {e}")
        finally:
            fila.task_done()


async def pipeline():
    urls = [f"https://httpbin.org/status/{code}"
            for code in [200, 201, 204, 400, 404, 500] * 2]

    fila = asyncio.Queue(maxsize=5)

    async with httpx.AsyncClient() as client:
        consumidores = [
            asyncio.create_task(consumidor(f"C{i}", fila, client))
            for i in range(1, N_CONSUMIDORES + 1)
        ]
        await produtor(fila, urls)
        await asyncio.gather(*consumidores)


asyncio.run(pipeline())

O número de sentinelas None tem de ser igual ao de consumidores, e por isso os dois saem da mesma constante. Com quatro consumidores e três sentinelas, o quarto fica parado no fila.get() para sempre: o programa não termina e não mostra erro nenhum.

Escolhendo a Abordagem Certa

# Guia de decisão rápida

def escolher_abordagem(tipo_tarefa, volume):
    if tipo_tarefa == "IO":
        if volume > 100:
            return "asyncio + httpx"         # máxima escalabilidade
        else:
            return "ThreadPoolExecutor"      # simples e eficiente
    elif tipo_tarefa == "CPU":
        return "ProcessPoolExecutor"         # paralelismo real
    elif tipo_tarefa == "misto":
        return "ProcessPoolExecutor + asyncio dentro de cada processo"

Exemplo Completo: Crawler Assíncrono

import asyncio
import re
import time
from dataclasses import dataclass, field
from urllib.parse import urljoin, urldefrag, urlparse

import httpx


RE_TITULO = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
RE_LINK   = re.compile(r"""<a\s[^>]*href=["']([^"'#]+)""", re.IGNORECASE)


@dataclass
class ResultadoCrawl:
    url:        str
    status:     int
    titulo:     str = ""
    links:      list = field(default_factory=list)
    erro:       str = ""


class CrawlerAssincrono:
    def __init__(self, max_paginas=10, max_workers=5):
        self.max_paginas  = max_paginas
        self.max_workers  = max_workers
        self._visitados:  set[str] = set()
        self._resultados: list = []
        self._semaforo    = None

    async def _processar_url(self, client, url):
        async with self._semaforo:
            try:
                r = await client.get(url, timeout=10.0, follow_redirects=True)
            except httpx.HTTPError as e:
                return ResultadoCrawl(url=url, status=0, erro=str(e))

            titulo, links = "", []
            if "text/html" in r.headers.get("content-type", ""):
                match  = RE_TITULO.search(r.text)
                titulo = match.group(1).strip() if match else ""
                # links absolutos, sem fragmento, só do mesmo domínio
                base = str(r.url)
                for href in RE_LINK.findall(r.text):
                    absoluto, _ = urldefrag(urljoin(base, href))
                    if urlparse(absoluto).netloc == urlparse(base).netloc:
                        links.append(absoluto)

            return ResultadoCrawl(
                url=url, status=r.status_code, titulo=titulo, links=links
            )

    async def crawlear(self, url_inicial: str):
        self._semaforo = asyncio.Semaphore(self.max_workers)
        fila = asyncio.Queue()
        await fila.put(url_inicial)
        self._visitados.add(url_inicial)

        async with httpx.AsyncClient(
            headers={"User-Agent": "PythonCrawler/1.0"}
        ) as client:
            while not fila.empty() and len(self._resultados) < self.max_paginas:
                vagas   = self.max_paginas - len(self._resultados)
                tarefas = []
                while not fila.empty() and len(tarefas) < min(self.max_workers, vagas):
                    url = await fila.get()
                    tarefas.append(self._processar_url(client, url))

                resultados = await asyncio.gather(*tarefas)
                self._resultados.extend(resultados)

                # o que torna isto um crawler: os links achados entram na fila
                for resultado in resultados:
                    for link in resultado.links:
                        if link not in self._visitados:
                            self._visitados.add(link)
                            await fila.put(link)

        return self._resultados


async def main():
    crawler = CrawlerAssincrono(max_paginas=5, max_workers=3)
    inicio  = time.perf_counter()

    resultados = await crawler.crawlear("https://www.python.org/")
    total      = time.perf_counter() - inicio

    print(f"\n=== Crawl concluído em {total:.2f}s ===")
    for r in resultados:
        if r.erro:
            print(f"  ✗ {r.url[:50]:50} — {r.erro[:30]}")
        else:
            print(f"  ✓ [{r.status}] {r.titulo[:40]:40} — {r.url[:40]}")


asyncio.run(main())

Duas decisões deste crawler valem para qualquer código de rede. A primeira: um link só entra na fila se ainda não estiver em _visitados, e é isso que impede o programa de girar em círculo entre páginas que se citam. A segunda: não há verify=False. Desligar a verificação do certificado faz o erro de TLS sumir, mas também aceita qualquer servidor que se passe pelo site — e, ao contrário do requests, que emite InsecureRequestWarning, o httpx não avisa nada. Para HTML de verdade, com marcação malformada, prefira um parser como o BeautifulSoup; a expressão regular aqui basta para o exemplo.

A pergunta que decide a ferramenta é onde o programa passa o tempo. Se ele espera — rede, disco, banco —, threads e asyncio resolvem, porque o GIL é liberado durante a espera. O ThreadPoolExecutor é o caminho mais curto para código que já usa bibliotecas síncronas como o requests, e ainda devolve as exceções das threads por future.result(). O asyncio escala para milhares de conexões num único thread, desde que tudo no caminho seja assíncrono; o gather não executa nada em paralelo, apenas deixa as corrotinas esperarem juntas. E toda fila com sentinelas precisa de uma para cada consumidor, ou o programa termina sem terminar.

Se o programa calcula, a resposta ainda é o processo. Um pool só entrega o ganho esperado quando as tarefas têm tamanho parecido e duram bem mais que o custo de criar um processo, e desde a 3.14 a guarda if __name__ == "__main__" é obrigatória também no Linux. O build sem GIL já permite threads em paralelo de verdade, ao custo de uma thread isolada mais lenta e de extensões em C que ainda estão chegando. Em qualquer dos modelos, estado compartilhado pede Lock — não porque o erro apareça no primeiro teste, mas justamente porque quase nunca aparece.

Fontes e leituras recomendadas

• asyncio — documentação oficial — https://docs.python.org/3/library/asyncio.html
• concurrent.futures — https://docs.python.org/3/library/concurrent.futures.html
• multiprocessing — https://docs.python.org/3/library/multiprocessing.html
• threading — https://docs.python.org/3/library/threading.html
• PEP 492 — corrotinas com async/await — https://peps.python.org/pep-0492/
• BEAZLEY, David. Python Concurrency from the Ground Up — palestra PyCon 2015, disponível em https://youtu.be/MCs5OvhV9S4
• RAMALHO, Luciano. Fluent Python. 2. ed. O'Reilly Media, 2022. Cap. 19–21 — concorrência, paralelismo e asyncio em profundidade.
• MATTHES, Eric. Python Crash Course. 3. ed. No Starch Press, 2023. — base para entender o modelo sequencial antes de migrar para concorrência.

Exercícios

Exercício 1

Um sistema conta os downloads de cada arquivo num dicionário compartilhado entre as threads do servidor. O teste automatizado dispara 10 threads com mil downloads cada e confere o total: passa sempre, em centenas de execuções. Em produção, o painel mostra bem menos downloads do que o log de acesso registra. A única diferença é que, em produção, o incremento ficou assim: atual = contagem[arquivo], depois uma linha de log gravada em disco, depois contagem[arquivo] = atual + 1. Explique por que o teste nunca pegou o problema e o que deveria estar no código.

Exercício 2

Um analista paraleliza um relatório com multiprocessing.Pool(processes=4) numa máquina de 16 núcleos. São quatro tarefas: contar os primos até 100 mil, 200 mil, 300 mil e 400 mil. O relatório sequencial levava 1,65 s; com o pool, 1,01 s. Ele conclui que o multiprocessing tem overhead demais e aumenta para 16 processos, sem ganho nenhum. O que ele deveria ter olhado, e como tirar mais do mesmo hardware?

Exercício 3

Um script de processamento de imagens roda há dois anos num servidor Linux com Python 3.12. Ele cria um multiprocessing.Pool direto no corpo do módulo, sem função main. Depois da atualização para a 3.14, o script passa a falhar logo no início com RuntimeError, e a mensagem fala em "bootstrapping phase". Ninguém mexeu no código. O que mudou, e qual é a correção certa?

Exercício 4

Um pipeline assíncrono tem um produtor e três consumidores ligados por uma asyncio.Queue, e o produtor encerra colocando três None na fila. Para dar conta do volume, alguém aumenta os consumidores para quatro. A partir daí o job noturno nunca termina: todos os itens aparecem processados no log, não há exceção, e o processo fica vivo até ser morto de manhã. Explique o travamento e duas formas de evitar que ele volte.

Exercício 5

Uma API assíncrona precisa gerar uma miniatura de imagem e chamar um serviço legado que só tem cliente síncrono. O desenvolvedor escreve as duas coisas dentro de async def, com requests.get(...) e a função de redimensionar, e dispara três de uma vez com asyncio.gather. Os testes passam, mas o tempo total é a soma das três chamadas, e durante esse intervalo a API não responde nem ao health check. Por que o gather não ajudou, e como corrigir cada uma das duas chamadas?