Noções Básicas de Concorrência

9 exemplos para você começar com Concorrência - 6 básicos e 3 intermediários.

Pré-requisitos

• Python 3.14.0 instalado.
• Leia Escolhendo um Modelo de Concorrência para orientação sobre I/O vs CPU.

Exemplos Básicos

1. Concorrência vs Paralelismo

Concorrência intercala tarefas; paralelismo executa tarefas simultaneamente em múltiplos núcleos.

# Concorrente: uma thread alterna enquanto outra espera por I/O
# Paralelo: dois processos em dois núcleos de CPU computam ao mesmo tempo

• Trabalho limitado por I/O se beneficia de concorrência (threads/async).
• Trabalho limitado por CPU precisa de paralelismo (processos) devido ao GIL.
• Async é concorrente, mas geralmente single-threaded.

Relacionado: O GIL e Python Livre de Threads - Detalhes do GIL

2. Thread para Espera de I/O

Threads sobrepõem I/O bloqueante.

import time
from threading import Thread
 
def fetch(_id: int) -> None:
    time.sleep(0.1)  # substituto para rede
    print("done", _id)
 
threads = [Thread(target=fetch, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

• sleep simula a liberação do GIL por I/O bloqueante.
• join garante que todo o trabalho seja concluído antes da saída.
• Threads compartilham memória - proteja estado mutável com locks.

Relacionado: threading - locks e segurança

3. Processo para Trabalho de CPU

Processos contornam o GIL com interpretadores separados.

from multiprocessing import Pool
 
def square(n: int) -> int:
    return n * n
 
if __name__ == "__main__":
    with Pool(2) as pool:
        print(pool.map(square, range(4)))

• Guarda if __name__ == "__main__" necessária em plataformas de spawn.
• Processos não compartilham memória por padrão.
• Custo de inicialização maior que threads - agrupe o trabalho.

Relacionado: multiprocessing - pools e IPC

4. ThreadPool do concurrent.futures

API de pool de alto nível para threads e processos.

from concurrent.futures import ThreadPoolExecutor, as_completed
 
def work(n: int) -> int:
    return n * 2
 
with ThreadPoolExecutor(max_workers=4) as ex:
    futures = [ex.submit(work, i) for i in range(6)]
    for fut in as_completed(futures):
        print(fut.result())

• submit retorna Future; map preserva a ordem.
• as_completed retorna à medida que cada um termina.
• O gerenciador de contexto encerra os workers de forma limpa.

Relacionado: concurrent.futures - API completa

5. Coordenação com queue.Queue

Produtor/consumidor desacopla estágios.

from queue import Queue
from threading import Thread
 
q: Queue[int] = Queue()
 
def producer() -> None:
    for i in range(3):
        q.put(i)
    q.put(-1)  # sentinela
 
def consumer() -> None:
    while True:
        item = q.get()
        if item == -1:
            break
        print("got", item)
 
Thread(target=producer).start()
Thread(target=consumer).start()

• Queue é thread-safe e bloqueante.
• Sentinela encerra o loop do consumidor de forma limpa.
• Prefira join() na fila para pipelines de produção.

Relacionado: Filas e Produtor/Consumidor - pipelines

6. Risco de Estado Compartilhado

Incrementos não protegidos causam race conditions.

import threading
 
counter = 0
 
def inc() -> None:
    global counter
    for _ in range(100_000):
        counter += 1
 
threads = [threading.Thread(target=inc) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)  # frequentemente < 200_000

• counter += 1 é uma operação de leitura-modificação-escrita - não é atômica no bytecode do Python.
• Use threading.Lock ou tipos atômicos para contadores compartilhados.
• Prefira passagem de mensagens em vez de estado mutável compartilhado.

Relacionado: Melhores Práticas de Concorrência - evite race conditions

Exemplos Intermediários

7. I/O Concorrente com asyncio

Loop de eventos single-threaded com muitas tarefas aguardando.

import asyncio
 
async def fetch(n: int) -> int:
    await asyncio.sleep(0.1)
    return n
 
async def main() -> None:
    results = await asyncio.gather(*(fetch(i) for i in range(3)))
    print(results)
 
asyncio.run(main())

• asyncio é adequado para muitas esperas de I/O concorrentes.
• Não paraleliza Python limitado por CPU em uma única thread.
• Veja a seção asyncio para padrões de produção.

Relacionado: Noções Básicas de Asyncio - loop de eventos

8. Preview de Subinterpretadores

Interpretadores isolados reduzem a contenção do GIL (APIs experimentais a partir do 3.12+).

# Conceitual - use o módulo interpreters quando habilitado para sua carga de trabalho
# Cada subinterpretador pode possuir um GIL em builds com threads livres
print("avaliar subinterpretadores para isolamento de CPU em 3.13+")

• Subinterpretadores visam paralelismo de CPU sem o overhead completo de processos.
• APIs evoluem - leia a página dedicada antes do uso em produção.
• Não é um substituto direto para todos os casos de multiprocessing.

Relacionado: Subinterpretadores - quando adotar

9. Checklist para Escolher o Modelo

Decisão rápida antes de importar uma biblioteca.

def model_hint(bound: str) -> str:
    if bound == "io":
        return "asyncio ou threads"
    if bound == "cpu":
        return "multiprocessing ou subprocess"
    return "profile primeiro"

• Faça profiling antes de mudar de modelo - suposições podem ser erradas.
• Cargas de trabalho mistas dividem a camada de I/O async + pool de processos para CPU.
• Documente a escolha em um ADR para alinhamento da equipe.

Relacionado: Escolhendo um Modelo de Concorrência - checklist completo

Versões da Stack: Esta página foi escrita para Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.