Multithreading no Python: threading, ThreadPoolExecutor e GIL

1
2
3
4
5
6
7
8
9

import threading, time

def baixar(arquivo):
    time.sleep(1)  # simula espera de rede
    print(f"{arquivo} pronto")

ts = [threading.Thread(target=baixar, args=(a,)) for a in ("a.csv", "b.csv")]
for t in ts: t.start()
for t in ts: t.join()  # 2 arquivos em ~1s, não em 2s

Salve salve Pythonista!

Se o seu script baixa arquivos um por um ou espera uma API responder a cada chamada, ele passa a maior parte do tempo parado. Multithreading resolve isso: enquanto uma thread espera, outra trabalha.

No post sobre concorrência e paralelismo em Python comparamos threads, processos e asyncio em alto nível. Aqui é o aprofundamento de threads: threading.Thread, ThreadPoolExecutor, condição de corrida com Lock, RLock, Event, Semaphore, Queue, o GIL com tempos medidos e o Python sem GIL. No final há erros comuns e exercícios resolvidos.

Os exemplos foram executados no Python 3.10 e no Python 3.14 (incluindo o build free-threaded 3.14t). Os tempos são reais, arredondados, medidos em um Ryzen 5 3600 (6 núcleos): na sua máquina vão variar, mas as proporções se mantêm.

Então… Bora pro post!

Vá Direto ao Assunto…

• O que é multithreading no Python
• Criando threads com threading.Thread
• ThreadPoolExecutor: o jeito moderno
• Condição de corrida e Lock
• Coordenando threads: Event, Semaphore e Queue
• GIL: quando threads aceleram e quando não
• Python sem GIL: o build free-threaded (3.13t e 3.14t)
• Threading, ThreadPoolExecutor, multiprocessing ou asyncio?
• Erros comuns
• Exercícios resolvidos
• Conclusão

O que é multithreading no Python

Uma thread é uma linha de execução dentro de um processo. Todo programa Python começa com uma, a MainThread. Multithreading é criar outras threads no mesmo processo, e todas enxergam as mesmas variáveis e objetos.

A vantagem: compartilhar dados é trivial e criar uma thread é bem mais barato que criar um processo. O risco: se duas threads alteram o mesmo dado ao mesmo tempo, o resultado pode sair errado (a condição de corrida, que veremos na prática).

No CPython padrão existe ainda o GIL (Global Interpreter Lock), que deixa apenas uma thread executar bytecode Python por vez. Uma thread que está esperando rede, disco ou time.sleep() libera o GIL, e é por isso que threads funcionam tão bem para tarefas I/O-bound (de espera). A documentação do módulo threading diz isso com todas as letras: threading “ainda é um modelo apropriado” para rodar várias tarefas de I/O simultaneamente.

Criando threads com threading.Thread

A forma básica é passar a função em target e os argumentos em args (sempre uma tupla). start() coloca a thread para rodar e join() faz a thread principal esperar por ela.

Vamos simular 4 downloads de 1 segundo cada, primeiro em sequência e depois com uma thread por arquivo:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import threading
import time

def baixar(arquivo):
    print(f"Baixando {arquivo}...")
    time.sleep(1)  # simula 1 segundo de espera por rede
    print(f"{arquivo} concluído")

arquivos = ["vendas.csv", "clientes.csv", "estoque.csv", "notas.csv"]

# 1) Sequencial
inicio = time.perf_counter()
for arquivo in arquivos:
    baixar(arquivo)
print(f"Sequencial: {time.perf_counter() - inicio:.2f}s\n")

# 2) Uma thread por arquivo
inicio = time.perf_counter()
threads = []
for arquivo in arquivos:
    t = threading.Thread(target=baixar, args=(arquivo,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()  # espera cada thread terminar
print(f"Com threads: {time.perf_counter() - inicio:.2f}s")

Saída (a ordem das linhas “concluído” muda a cada execução):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

Baixando vendas.csv...
vendas.csv concluído
Baixando clientes.csv...
clientes.csv concluído
Baixando estoque.csv...
estoque.csv concluído
Baixando notas.csv...
notas.csv concluído
Sequencial: 4.00s

Baixando vendas.csv...
Baixando clientes.csv...
Baixando estoque.csv...
Baixando notas.csv...
estoque.csv concluído
clientes.csv concluído
notas.csv concluído
vendas.csv concluído
Com threads: 1.00s

As 4 esperas aconteceram ao mesmo tempo, então o total caiu de 4 para 1 segundo. Repare em dois detalhes:

• Os start() vêm todos antes dos join(). Se você fizer t.start(); t.join() dentro do mesmo loop, cada thread termina antes da próxima começar e volta tudo a ser sequencial.
• A ordem de execução entre threads não é garantida. Nunca escreva código que dependa dela.

A classe também aceita kwargs (dicionário de argumentos nomeados), name (nome da thread, útil em logs) e daemon. Se *args e **kwargs ainda confundem, veja o post sobre args e kwargs no Python.

Threads daemon

Por padrão, o programa só termina quando todas as threads terminam. Uma thread daemon é diferente: ela é encerrada automaticamente quando a thread principal acaba. É útil para tarefas de fundo que não precisam de fim limpo, como um monitor:

1
2
3
4
5
6
7
8
9
10
11
12
13

import threading
import time

def monitorar():
    while True:
        print("monitorando...")
        time.sleep(0.4)

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

time.sleep(1)  # o programa principal faz seu trabalho
print("Programa principal terminou")

1
2
3
4

monitorando...
monitorando...
monitorando...
Programa principal terminou

Sem daemon=True, esse programa rodaria para sempre. Cuidado: a documentação avisa que threads daemon são interrompidas abruptamente no encerramento, e arquivos abertos ou transações podem ficar pela metade. Para parar uma thread com segurança, use um Event (mais abaixo). Os antigos setDaemon() e isDaemon() estão obsoletos desde o Python 3.10: use o atributo daemon.

Subclasse de Thread

Outra forma é herdar de threading.Thread e sobrescrever o método run(). Ela é útil quando a thread tem estado próprio, por exemplo para guardar o resultado, já que Thread descarta o valor retornado pela função alvo:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import threading
import time

class Download(threading.Thread):
    def __init__(self, arquivo, segundos):
        super().__init__(name=f"dl-{arquivo}")
        self.arquivo = arquivo
        self.segundos = segundos
        self.resultado = None  # a thread guarda o resultado aqui

    def run(self):
        time.sleep(self.segundos)  # simula I/O
        self.resultado = f"{self.arquivo} baixado em {self.segundos}s"

downloads = [Download("a.csv", 0.3), Download("b.csv", 0.1)]
for d in downloads:
    d.start()
for d in downloads:
    d.join()
    print(d.name, "->", d.resultado)

1
2

dl-a.csv -> a.csv baixado em 0.3s
dl-b.csv -> b.csv baixado em 0.1s

Você continua chamando start(), nunca run(). O super().__init__() é explicado no post sobre herança no Python.

ThreadPoolExecutor: o jeito moderno

Gerenciar threads na mão não escala: 1.000 URLs viram 1.000 threads. O concurrent.futures.ThreadPoolExecutor usa um pool: um número fixo de threads (max_workers) que pegam as tarefas de uma fila interna. Ele também devolve resultados e exceções, coisa que Thread não faz.

Se você não passar max_workers, o padrão é min(32, os.cpu_count() + 4) desde o Python 3.8 (no 3.13 passou a usar os.process_cpu_count()).

map(): mesma função para vários itens

Para ter saídas reproduzíveis, em vez de acessar sites externos, o exemplo abaixo sobe um servidor HTTP local que demora 0,5 segundo por resposta. O servidor roda em uma thread daemon e usa ThreadingHTTPServer, que atende cada requisição em uma thread própria:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

import threading
import time
from concurrent.futures import ThreadPoolExecutor
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.request import urlopen

class ServidorLento(BaseHTTPRequestHandler):
    def do_GET(self):
        time.sleep(0.5)                       # cada resposta demora 0,5s
        corpo = f"conteudo de {self.path}".encode()
        self.send_response(200)
        self.send_header("Content-Length", str(len(corpo)))
        self.end_headers()
        self.wfile.write(corpo)

    def log_message(self, *args):             # silencia o log
        pass

servidor = ThreadingHTTPServer(("127.0.0.1", 8000), ServidorLento)
threading.Thread(target=servidor.serve_forever, daemon=True).start()

URLS = [f"http://127.0.0.1:8000/pagina/{i}" for i in range(10)]

def baixar(url):
    with urlopen(url, timeout=5) as resposta:
        return len(resposta.read())

inicio = time.perf_counter()
tamanhos = [baixar(url) for url in URLS]
print(f"Sequencial: {time.perf_counter() - inicio:.2f}s")

inicio = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as executor:
    tamanhos = list(executor.map(baixar, URLS))
print(f"5 threads: {time.perf_counter() - inicio:.2f}s")
print(tamanhos)

servidor.shutdown()

1
2
3

Sequencial: 5.02s
5 threads: 1.01s
[21, 21, 21, 21, 21, 21, 21, 21, 21, 21]

Dez requisições de 0,5s em sequência levam 5s; com 5 threads, são duas “ondas”: 1s. Três pontos importantes:

• executor.map() devolve os resultados na ordem da entrada, mesmo que as tarefas terminem fora de ordem.
• O bloco with chama shutdown(wait=True) ao sair, ou seja, espera todas as tarefas terminarem.
• Em projetos reais, troque urlopen por requests em Python: o padrão é o mesmo.

submit() e as_completed(): resultados conforme ficam prontos

submit() agenda uma chamada e devolve um Future, um objeto que representa um resultado que ainda vai existir. as_completed() entrega os futures na ordem em que terminam, o que permite processar o primeiro resultado sem esperar o mais lento. E aqui está o tratamento de exceção: se a função falhou dentro da thread, future.result() relança a mesma exceção na thread principal.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def consultar(cep):
    time.sleep(len(cep) / 10)  # tempos diferentes por tarefa
    if not cep.isdigit():
        raise ValueError(f"CEP inválido: {cep!r}")
    return f"{cep} -> ok"

ceps = ["70000000", "0100", "abc12", "22040"]

with ThreadPoolExecutor(max_workers=4) as executor:
    futuros = {executor.submit(consultar, cep): cep for cep in ceps}

    for futuro in as_completed(futuros):
        cep = futuros[futuro]
        try:
            print(futuro.result())  # relança a exceção da thread, se houver
        except ValueError as erro:
            print(f"Falhou ({cep}): {erro}")

1
2
3
4

0100 -> ok
22040 -> ok
Falhou (abc12): CEP inválido: 'abc12'
70000000 -> ok

O CEP mais curto terminou primeiro, e a falha de um item não derrubou os outros. O dicionário futuros é um truque comum para saber a qual entrada cada resultado pertence. Com map(), a exceção aparece quando você chega ao item que falhou ao percorrer o iterador; os itens anteriores já foram entregues. Para revisar try/except, veja o post sobre tratamento de exceções no Python.

Está curtindo esse conteúdo?

Que tal receber 30 dias de conteúdo direto na sua Caixa de Entrada?

Sua assinatura não pôde ser validada.

Você fez sua assinatura com sucesso.

Assine as PyDicas e receba 30 dias do melhor conteúdo Python na sua Caixa de Entrada: direto e sem enrolação!

Seu nome

Seu e-mail

ASSINAR PYDICAS

Condição de corrida e Lock

Aqui mora o maior perigo das threads. Quatro threads fazem 1.000 depósitos cada em um saldo compartilhado. A operação “ler o saldo, somar 1, gravar” parece atômica, mas não é. O time.sleep(0) apenas deixa explícito o ponto em que o sistema pode trocar de thread no meio da operação:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import threading
import time

saldo = 0

def depositar(vezes):
    global saldo
    for _ in range(vezes):
        atual = saldo       # 1. lê
        time.sleep(0)       # 2. cede a vez (simula uma pausa qualquer)
        saldo = atual + 1   # 3. escreve

threads = [threading.Thread(target=depositar, args=(1000,)) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Esperado: 4000, obtido: {saldo}")

Três execuções seguidas no Python 3.10:

1
2
3

Esperado: 4000, obtido: 1050
Esperado: 4000, obtido: 1000
Esperado: 4000, obtido: 1037

Três quartos dos depósitos sumiram: duas threads leem o mesmo saldo (digamos 500), as duas gravam 501, e um depósito se perde. Isso é uma condição de corrida (race condition): o resultado depende da ordem em que as threads são intercaladas.

“Mas sem o sleep(0) funciona”, você pode dizer. No CPython com GIL, um simples saldo += 1 com 4 threads de 1 milhão de iterações deu o valor certo nos nossos testes (3.10 e 3.14). No build sem GIL (python3.14t), o mesmo código deu obtido: 1283498 de 4.000.000. O GIL nunca foi uma garantia para o seu código: ele só torna o bug mais raro e mais difícil de reproduzir.

A correção é um threading.Lock: só uma thread por vez entra no trecho protegido (a seção crítica). Use sempre com with, que libera o lock mesmo se ocorrer uma exceção:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

import threading
import time

saldo = 0
lock = threading.Lock()

def depositar(vezes):
    global saldo
    for _ in range(vezes):
        with lock:              # só uma thread por vez entra aqui
            atual = saldo
            time.sleep(0)
            saldo = atual + 1

threads = [threading.Thread(target=depositar, args=(1000,)) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Esperado: 4000, obtido: {saldo}")

1

Esperado: 4000, obtido: 4000

O with lock: equivale a acquire() com try/finally e release(), sem a chance de esquecer a liberação. Mantenha a seção crítica pequena: tudo dentro dela roda uma thread de cada vez.

RLock: quando a mesma thread precisa do lock de novo

Um Lock comum não sabe quem o adquiriu. Se a mesma thread tentar adquiri-lo duas vezes, ela fica esperando por si mesma para sempre (deadlock). Isso acontece quando um método protegido chama outro método protegido pelo mesmo lock. O RLock (lock reentrante) permite que a mesma thread o adquira várias vezes, desde que libere o mesmo número de vezes:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import threading

class Conta:
    def __init__(self, trava):
        self.saldo = 0
        self.trava = trava

    def depositar(self, valor):
        with self.trava:
            self.saldo += valor

    def depositar_com_bonus(self, valor):
        with self.trava:                 # 1ª aquisição
            self.depositar(valor)        # 2ª aquisição, pela MESMA thread
            self.depositar(valor * 0.1)

conta = Conta(threading.RLock())
conta.depositar_com_bonus(100)
print(conta.saldo)

# Com Lock comum, a mesma thread trava esperando por ela mesma:
trava = threading.Lock()
trava.acquire()
print(trava.acquire(timeout=1))  # False: com Lock() sem timeout, travaria para sempre

1
2

110.0
False

Com Conta(threading.Lock()), a chamada depositar_com_bonus(100) nunca terminaria. Regra prática: use Lock por padrão (é mais rápido e denuncia desenho confuso) e RLock quando houver aquisição aninhada que você não consegue evitar.

Coordenando threads: Event, Semaphore e Queue

Lock protege dados. Para coordenar threads (avisar, limitar, passar trabalho), há outras primitivas.

Event: sinalizar e parar threads com segurança

Um Event é uma flag compartilhada: set() liga, clear() desliga, is_set() consulta e wait() bloqueia até ela ser ligada (ou até o timeout). É a forma recomendada de pedir para uma thread parar:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import threading
import time

parar = threading.Event()

def trabalhador():
    ciclos = 0
    while not parar.is_set():
        ciclos += 1
        parar.wait(0.2)  # dorme 0,2s, mas acorda na hora se o evento for setado
    print(f"Trabalhador encerrado com segurança após {ciclos} ciclos")

t = threading.Thread(target=trabalhador)
t.start()
time.sleep(1)
parar.set()   # pede para a thread parar
t.join()
print("Fim")

1
2

Trabalhador encerrado com segurança após 5 ciclos
Fim

O número de ciclos pode variar em um de uma execução para outra. Usar parar.wait(0.2) no lugar de time.sleep(0.2) faz a thread reagir ao pedido de parada na hora. Python não tem como “matar” uma thread por fora: ela precisa sair sozinha, e o Event é o sinal.

Semaphore: limitar a concorrência

Um Semaphore(n) deixa no máximo n threads entrarem em um trecho ao mesmo tempo. É o que você usa para não estourar o limite de conexões de uma API ou de um banco:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import threading
import time

limite = threading.Semaphore(3)   # no máximo 3 ao mesmo tempo
ativos = 0
pico = 0
contador_lock = threading.Lock()

def chamar_api():
    global ativos, pico
    with limite:
        with contador_lock:
            ativos += 1
            pico = max(pico, ativos)
        time.sleep(0.2)            # simula a chamada
        with contador_lock:
            ativos -= 1

inicio = time.perf_counter()
threads = [threading.Thread(target=chamar_api) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"Pico de chamadas simultâneas: {pico}")
print(f"Tempo total: {time.perf_counter() - inicio:.1f}s")

1
2

Pico de chamadas simultâneas: 3
Tempo total: 0.8s

Dez chamadas, três por vez, 0,2s cada: quatro ondas, 0,8s. O BoundedSemaphore funciona igual, mas gera ValueError se você liberar mais vezes do que adquiriu, o que ajuda a achar bugs. Se todas as tarefas passam por um ThreadPoolExecutor, o próprio max_workers já faz esse papel; o semáforo brilha quando só uma parte do código precisa de limite.

queue.Queue: produtor e consumidor

A forma mais segura de trocar dados entre threads é não compartilhar variáveis: use uma queue.Queue, que já faz todo o controle de lock internamente. put() coloca um item (e bloqueia se a fila tiver maxsize e estiver cheia), get() retira (e bloqueia até haver item), task_done() avisa que o item foi processado e join() espera todos serem processados:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import queue
import threading
import time

fila = queue.Queue(maxsize=5)   # put() bloqueia quando há 5 itens
FIM = None                      # sentinela: "acabou o trabalho"

def consumidor(nome):
    while True:
        item = fila.get()       # bloqueia até chegar um item
        if item is FIM:
            fila.task_done()
            break
        time.sleep(0.1)         # simula o processamento
        print(f"{nome} processou {item}")
        fila.task_done()

consumidores = [threading.Thread(target=consumidor, args=(f"C{i}",)) for i in (1, 2)]
for c in consumidores:
    c.start()

for i in range(1, 5):           # a thread principal é o produtor
    fila.put(f"pedido-{i}")
for _ in consumidores:
    fila.put(FIM)               # uma sentinela por consumidor

fila.join()                     # espera todo item receber task_done()
print("Todos os pedidos processados")

1
2
3
4
5

C1 processou pedido-1
C2 processou pedido-2
C1 processou pedido-3
C2 processou pedido-4
Todos os pedidos processados

A sentinela (None) é o jeito clássico de avisar cada consumidor que não vem mais nada. Filas FIFO, LIFO, de prioridade e mais detalhes do produtor-consumidor estão no post sobre como gerenciar filas em Python com a biblioteca queue.

GIL: quando threads aceleram e quando não

Até aqui, todas as tarefas esperavam (sleep, rede). E se a tarefa for cálculo puro? Vamos contar primos até 200.000 quatro vezes: em sequência, com 4 threads e com 4 processos (ProcessPoolExecutor, que tem a mesma interface do pool de threads):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import sys
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

def contar_primos(limite):
    total = 0
    for n in range(2, limite):
        if all(n % d for d in range(2, int(n ** 0.5) + 1)):
            total += 1
    return total

TAREFAS = [200_000] * 4

def medir(nome, funcao):
    inicio = time.perf_counter()
    funcao()
    print(f"{nome:<16} {time.perf_counter() - inicio:.2f}s")

if __name__ == "__main__":
    gil = sys._is_gil_enabled() if hasattr(sys, "_is_gil_enabled") else True
    print(f"Python {sys.version.split()[0]}, GIL ativo: {gil}")
    medir("Sequencial", lambda: [contar_primos(n) for n in TAREFAS])
    with ThreadPoolExecutor(max_workers=4) as ex:
        medir("4 threads", lambda: list(ex.map(contar_primos, TAREFAS)))
    with ProcessPoolExecutor(max_workers=4) as ex:
        medir("4 processos", lambda: list(ex.map(contar_primos, TAREFAS)))

1
2
3
4

Python 3.10.12, GIL ativo: True
Sequencial       1.60s
4 threads        2.34s
4 processos      0.45s

Com GIL, as 4 threads foram mais lentas que o sequencial: só uma executa por vez, e alternar entre elas custa. Os 4 processos, cada um com seu GIL, usaram 4 núcleos e foram cerca de 3,5 vezes mais rápidos. No Python 3.14 com GIL o comportamento se repetiu (1,29s, 2,14s e 0,40s). O if __name__ == "__main__": é obrigatório com processos, porque cada processo filho importa o seu script.

Resumindo a regra:

• I/O-bound (rede, disco, banco, sleep): threads aceleram, porque quem espera libera o GIL.
• CPU-bound (cálculo em Python puro): threads não ajudam no CPython com GIL. Use processos.
• Bibliotecas em C como NumPy costumam liberar o GIL em operações pesadas, e aí threads podem ganhar com CPU. Meça antes de concluir.

Python sem GIL: o build free-threaded (3.13t e 3.14t)

A PEP 703 tornou o GIL opcional no CPython. Conferimos o status na documentação oficial (guia de free-threading e What’s New do 3.14):

• No Python 3.13, o build free-threaded chegou como experimental, em um executável separado, normalmente python3.13t.
• No Python 3.14, ele passou a ser oficialmente suportado (PEP 779), mas continua opcional: o python3 padrão ainda tem GIL. A perda de desempenho em código de uma thread só caiu para cerca de 5% a 10%.
• Extensões em C precisam declarar suporte. Importar uma que não declara reativa o GIL, com um aviso.

Para descobrir em qual modo você está:

1
2
3
4
5
6

import sys  # requer Python 3.13+
import sysconfig

print(sys.version.split()[0])
print("Build free-threaded:", bool(sysconfig.get_config_var("Py_GIL_DISABLED")))
print("GIL ativo agora:", sys._is_gil_enabled())

Com python3.14 e com python3.14t, respectivamente:

1
2
3
4
5
6
7

3.14.3
Build free-threaded: False
GIL ativo agora: True

3.14.3
Build free-threaded: True
GIL ativo agora: False

E o benchmark de primos, rodando no python3.14t:

1
2
3
4

Python 3.14.3, GIL ativo: False
Sequencial       1.41s
4 threads        0.41s
4 processos      0.45s

Agora as threads rodam em paralelo de verdade e empatam com os processos, sem o custo de criá-los nem de serializar dados. O preço: toda condição de corrida vira realidade, como vimos com o saldo += 1, e locks passam a ser obrigatórios. Com -X gil=1 ou a variável de ambiente PYTHON_GIL=1 você religa o GIL no build free-threaded para comparar. Instalação e outros detalhes estão no post sobre as novidades do Python 3.13 e 3.14.

Escolher entre threading, multiprocessing e asyncio importa menos do que ter fundamentos de Python bem firmes - é essa base que a Jornada Python constrói com você, do zero ao avançado:

Domine Python: A Linguagem da Inteligência Artificial, do Desenvolvimento Web e das Carreiras do Futuro.

A mais completa formação para você se tornar um especialista em Python, pronto para construir desde aplicações web robustas com Django até projetos de Inteligência Artificial e Ciência de Dados .

Quero conhecer!

Nota 4,9 de 5 na Hotmart O curso de Python mais bem avaliado da Hotmart!

Threading, ThreadPoolExecutor, multiprocessing ou asyncio?

Se o seu caso é I/O com muitas conexões e você pode usar bibliotecas assíncronas, vale conhecer a programação assíncrona no Python. Threads têm a vantagem de funcionar com qualquer biblioteca síncrona que você já usa, sem reescrever o código com async/await.

Erros comuns

args sem vírgula (TypeError)

1
2
3
4
5
6
7
8

import threading

def baixar(arquivo):
    print(f"Baixando {arquivo}")

t = threading.Thread(target=baixar, args=("vendas.csv"))
t.start()
t.join()

1
2
3
4

Exception in thread Thread-1 (baixar):
Traceback (most recent call last):
  ...
TypeError: baixar() takes 1 positional argument but 10 were given

("vendas.csv") é só uma string entre parênteses, e o Python a desempacota letra por letra (10 caracteres, 10 argumentos). Correção: args=("vendas.csv",), com a vírgula que cria a tupla. Repare que o erro aparece dentro da thread e o programa principal segue em frente.

Chamar a função em target (TypeError)

1
2
3
4
5
6
7
8

import threading

def baixar(arquivo):
    return f"{arquivo} ok"

t = threading.Thread(target=baixar("vendas.csv"))
t.start()
t.join()

1
2
3
4

Exception in thread Thread-1:
Traceback (most recent call last):
  ...
TypeError: 'str' object is not callable

Com os parênteses, baixar roda na thread principal e o que vai para target é o retorno (uma string). Se a função retornar None, nem há erro: a thread só não faz nada. Correção: target=baixar, args=("vendas.csv",).

Iniciar a mesma thread duas vezes (RuntimeError)

1
2
3
4
5
6

import threading

t = threading.Thread(target=print, args=("oi",))
t.start()
t.join()
t.start()

1
2
3
4
5

oi
Traceback (most recent call last):
  File "erro.py", line 6, in <module>
    t.start()
RuntimeError: threads can only be started once

Um objeto Thread não é reutilizável. Correção: crie um objeto novo para cada execução, ou use um ThreadPoolExecutor, que reaproveita as threads por você.

Exceção “sumida” em um Future (ZeroDivisionError)

1
2
3
4
5
6
7
8
9
10

from concurrent.futures import ThreadPoolExecutor

def dividir(a, b):
    return a / b

with ThreadPoolExecutor() as executor:
    futuro = executor.submit(dividir, 10, 0)

print("Terminou sem erro?")
print(futuro.result())

1
2
3
4

Terminou sem erro?
Traceback (most recent call last):
  ...
ZeroDivisionError: division by zero

A exceção fica guardada no Future. Se você nunca chamar result() (ou exception()), o erro passa em silêncio. Correção: sempre leia futuro.result() dentro de um try/except, como no exemplo com as_completed().

Usar o executor depois do with (RuntimeError)

1
2
3
4
5
6

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor() as executor:
    executor.submit(print, "primeira")

executor.submit(print, "segunda")

1
2
3
4
5

primeira
Traceback (most recent call last):
  File "erro.py", line 6, in <module>
    executor.submit(print, "segunda")
RuntimeError: cannot schedule new futures after shutdown

Ao sair do with, o executor é desligado (shutdown()). Correção: agende todas as tarefas dentro do bloco with.

Exercícios resolvidos

Tente resolver cada um antes de abrir a solução. Todos foram executados e as saídas conferidas no Python 3.10. Para treinar a base, use também a nossa página de exercícios de Python resolvidos.

Exercício 1. Crie 3 threads, cada uma adicionando a frase "Olá da thread N" (N de 1 a 3) a uma lista compartilhada. Depois de todas terminarem, imprima a lista ordenada.

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import threading

mensagens = []

def saudar(numero):
    mensagens.append(f"Olá da thread {numero}")  # list.append é thread-safe

threads = [threading.Thread(target=saudar, args=(n,)) for n in range(1, 4)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(sorted(mensagens))

Exercício 2. Use ThreadPoolExecutor.map() para calcular o quadrado de 1 a 5, com uma função que dorme 0.5 - n / 10 segundos (os maiores terminam primeiro). Em que ordem os resultados aparecem?

1
2
3
4
5
6
7
8
9

import time
from concurrent.futures import ThreadPoolExecutor

def quadrado(n):
    time.sleep(0.5 - n / 10)  # os maiores terminam primeiro
    return n * n

with ThreadPoolExecutor(max_workers=5) as executor:
    print(list(executor.map(quadrado, range(1, 6))))

Exercício 3. Seis tarefas dormem 0,5s cada. Meça o tempo em sequência, com max_workers=2 e com max_workers=6.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import time
from concurrent.futures import ThreadPoolExecutor

def tarefa(_):
    time.sleep(0.5)

inicio = time.perf_counter()
for i in range(6):
    tarefa(i)
print(f"Sequencial: {time.perf_counter() - inicio:.1f}s")

for workers in (2, 6):
    inicio = time.perf_counter()
    with ThreadPoolExecutor(max_workers=workers) as executor:
        list(executor.map(tarefa, range(6)))
    print(f"{workers} workers: {time.perf_counter() - inicio:.1f}s")

1
2
3

Sequencial: 3.0s
2 workers: 1.5s
6 workers: 0.5s

Exercício 4. Crie uma classe ContadorSeguro com um método incrementar() protegido por lock. Oito threads chamam incrementar() 10.000 vezes cada. Imprima o valor final.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import threading

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

    def incrementar(self):
        with self._lock:
            self.valor += 1

contador = ContadorSeguro()

def trabalhar():
    for _ in range(10_000):
        contador.incrementar()

threads = [threading.Thread(target=trabalhar) for _ in range(8)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(contador.valor)

Exercício 5. Faça uma thread “corredor” esperar um sinal da thread principal para largar, usando threading.Event.

1
2
3
4
5
6
7
8
9
10
11
12
13

import threading

pronto = threading.Event()

def corredor(nome):
    pronto.wait()               # espera o sinal
    print(f"{nome} largou")

t = threading.Thread(target=corredor, args=("Ana",))
t.start()
print("Preparar... já!")
pronto.set()
t.join()

1
2

Preparar... já!
Ana largou

Exercício 6 (estilo prova). O que o código abaixo imprime?

1
2
3
4
5
6
7
8
9
10
11

import threading

def quem():
    print(threading.current_thread().name)

a = threading.Thread(target=quem, name="Trabalhador")
a.run()

b = threading.Thread(target=quem, name="Trabalhador")
b.start()
b.join()

a) Trabalhador e Trabalhador
b) MainThread e Trabalhador
c) MainThread e MainThread
d) RuntimeError

1
2

MainThread
Trabalhador

Exercício 7 (estilo prova). Um programa calcula números primos (só cálculo, em Python puro) e leva 8 segundos. O programador divide o trabalho em 4 threads no CPython 3.12 padrão, em uma máquina com 8 núcleos. O que é mais provável?

a) Cai para cerca de 2 segundos, pois cada thread usa um núcleo.
b) Fica em torno de 8 segundos ou mais, por causa do GIL.
c) O programa gera erro, pois threads não podem fazer cálculos.
d) Cai para 1 segundo, pois há 8 núcleos.

1
2

import sys
print(getattr(sys, "_is_gil_enabled", lambda: True)())  # True no CPython padrão

Exercício 8. Quatro threads calculam o cubo de 1, 2, 3 e 4 e colocam o par (n, cubo) em uma queue.Queue. Monte um dicionário ordenado com os resultados.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import queue
import threading

resultados = queue.Queue()

def calcular(n):
    resultados.put((n, n ** 3))

threads = [threading.Thread(target=calcular, args=(n,)) for n in range(1, 5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

cubos = dict(sorted(resultados.get() for _ in range(resultados.qsize())))
print(cubos)

Conclusão

Threads são a ferramenta certa para tarefas que esperam. Comece pelo ThreadPoolExecutor, sempre leia future.result(), proteja dado compartilhado com Lock (ou troque-o por uma Queue) e, para cálculo pesado, use processos ou o build free-threaded.

Quer firmar a base de Python por trás disso e construir APIs e automações com projetos guiados, do básico ao avançado? Conheça o nosso curso de Python completo.

Até a próxima, Pythonista!

Por Vinícius Ramos em 03/08/2026

Engenheiro de Computação pela UnB com especialização em Inteligência Artificial. Sr. Software Engineer com 10+ anos de experiência desenvolvendo sistemas críticos em Python e Django. Atuei no Banco do Brasil liderando equipes para monitoramento de mainframe, trabalhei como desenvolvedor sênior em projetos forenses para a Polícia Federal, e atualmente sou engenheiro de software na Labcorp (Estados Unidos), desenvolvendo soluções escaláveis para testes genéticos. Fundador da Python Academy, compartilhando conhecimento técnico com a comunidade brasileira desde 2018.

Bora se conectar no LinkedIn! vinicius-aramos

Minhas redes:

• 
• 
•