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.

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()
    resposta = requests.get(url, timeout=10)
    duracao  = time.perf_counter() - inicio
    resultados[indice] = {
        "url":    url,
        "status": resposta.status_code,
        "bytes":  len(resposta.content),
        "tempo":  duracao
    }


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

# Sequencial — ~4 segundos
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:
    print(f"  {r['status']} — {r['bytes']} bytes em {r['tempo']:.2f}s")

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

Sem o Lock, o resultado seria imprevisível — condição de corrida clássica.

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
    benchmark()

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


numeros = [
    999_999_937,
    998_244_353,
    1_000_000_007,
    999_999_893,
]

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))}")

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 em paralelo
async def demo_gather():
    inicio     = asyncio.get_event_loop().time()
    resultados = await asyncio.gather(
        tarefa("A", 1.0),
        tarefa("B", 1.5),
        tarefa("C", 0.5),
    )
    total = asyncio.get_event_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 asyncio.TimeoutError:
        print("Operação cancelada por timeout.")


asyncio.run(demo_gather())

Queue Assíncrona: Produtor e Consumidor

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

import asyncio
import httpx
import random


async def produtor(fila: asyncio.Queue, urls: list):
    for url in urls:
        await fila.put(url)
        print(f"  [Produtor] enfileirou: {url}")
    # Sinal de fim para cada consumidor
    for _ in range(3):
        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, 4)
        ]
        await produtor(fila, urls)
        await asyncio.gather(*consumidores)


asyncio.run(pipeline())

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"

Abordagem	Melhor para	Limitação
`threading`	I/O simples, legado	GIL, overhead de threads
`ThreadPoolExecutor`	I/O com interface limpa	GIL
`multiprocessing`	CPU intensivo	Overhead de processos, IPC
`ProcessPoolExecutor`	CPU com interface limpa	Overhead de processos
`asyncio`	I/O massivo, alta escala	Curva de aprendizado, single-thread

Exemplo Completo: Crawler Assíncrono

import asyncio
import httpx
from dataclasses import dataclass, field
from typing import Set
from urllib.parse import urljoin, urlparse
import time


@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)
                titulo = ""
                if "text/html" in r.headers.get("content-type", ""):
                    import re
                    match = re.search(r"<title>(.*?)</title>", r.text, re.IGNORECASE)
                    titulo = match.group(1).strip() if match else ""

                return ResultadoCrawl(
                    url=url, status=r.status_code, titulo=titulo
                )
            except Exception as e:
                return ResultadoCrawl(url=url, status=0, erro=str(e))

    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"},
            verify=False
        ) as client:
            while not fila.empty() and len(self._resultados) < self.max_paginas:
                tarefas = []
                while not fila.empty() and len(tarefas) < self.max_workers:
                    url = await fila.get()
                    tarefas.append(self._processar_url(client, url))

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

        return self._resultados


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

    resultados = await crawler.crawlear("https://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())

Resumo

GIL limita threads a concorrência, não paralelismo — para CPU use multiprocessing
ThreadPoolExecutor é a forma moderna de usar threads — simples e segura
multiprocessing.Pool e ProcessPoolExecutor entregam paralelismo real em múltiplos núcleos
asyncio é single-thread mas altamente escalável para I/O — usa corrotinas cooperativas
await cede o controle do event loop; asyncio.gather() executa corrotinas em paralelo
asyncio.Semaphore limita a concorrência — evita sobrecarga de servidores externos
asyncio.Queue implementa o padrão produtor-consumidor de forma elegante
Escolha a abordagem pelo tipo de gargalo: I/O → asyncio/threads, CPU → processos

Referências e Leituras Complementares

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.

Concorrência: threads, processos e async/await Já leu