Arquivo de Paralela ou Concorrente - Página 2 de 3

Multiprocessamento em Python: Paralelismo Real

22/05/202622/04/2026 Por antonino

Multiprocessamento cria múltiplos processos independentes. Cada processo tem seu próprio interpretador Python e memória. Primeiramente, isso contorna a limitação do GIL completamente. Por exemplo, 4 processos podem executar em 4 núcleos de CPU simultaneamente. Além disso, processos não compartilham memória por padrão. A voz passiva é usada aqui: “dados são copiados entre processos via serialização”. Quando utilizar multiprocessing? Em tarefas com CPU intensiva. Por exemplo, processamento de imagens, cálculos matemáticos ou criptografia. Também é útil para explorar todo o potencial da máquina. Python oferece o módulo multiprocessing para isso. Vamos explorar criação, comunicação e boas práticas. Três subtítulos guiarão você pelo paralelismo real. Ao final, você dominará o multiprocessamento em Python.

Criando e gerenciando processos

O módulo multiprocessing tem interface similar ao threading. Use Process para criar processos individuais. O método start() inicia o processo e join() espera terminar. Para múltiplas tarefas, use Pool para gerenciar um conjunto de processos. Quando usar criação manual? Em poucos processos com lógica específica. A voz passiva é aplicada: “os argumentos são passados via args“. Exemplo básico de criação de processos:

import multiprocessing
import time
import os

def trabalhador(nome, segundos):
    """Função executada por cada processo."""
    pid = os.getpid()
    print(f"Processo {nome} (PID: {pid}) iniciando")
    time.sleep(segundos)
    print(f"Processo {nome} finalizado após {segundos}s")
    return f"Resultado de {nome}"

if __name__ == "__main__":
    print(f"Processo principal PID: {os.getpid()}")
    
    # Criando processos manualmente
    p1 = multiprocessing.Process(target=trabalhador, args=("A", 2))
    p2 = multiprocessing.Process(target=trabalhador, args=("B", 1))
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()
    
    # Usando Pool para múltiplos processos
    print("\n=== Usando Pool ===")
    with multiprocessing.Pool(processes=4) as pool:
        # Mapear uma lista de argumentos para a função
        args_lista = [(f"Task-{i}", i % 3 + 1) for i in range(6)]
        resultados = pool.starmap(trabalhador, args_lista)
        print(f"Resultados: {resultados}")

import multiprocessing

import time

import os

def trabalhador(nome, segundos):

"""Função executada por cada processo."""

pid = os.getpid()

print(f"Processo {nome} (PID: {pid}) iniciando")

time.sleep(segundos)

print(f"Processo {nome} finalizado após {segundos}s")

return f"Resultado de {nome}"

if __name__ == "__main__":

print(f"Processo principal PID: {os.getpid()}")

# Criando processos manualmente

p1 = multiprocessing.Process(target=trabalhador, args=("A", 2))

p2 = multiprocessing.Process(target=trabalhador, args=("B", 1))

p1.start()

p2.start()

p1.join()

p2.join()

# Usando Pool para múltiplos processos

print("\n=== Usando Pool ===")

with multiprocessing.Pool(processes=4) as pool:

# Mapear uma lista de argumentos para a função

args_lista = [(f"Task-{i}", i % 3 + 1) for i in range(6)]

resultados = pool.starmap(trabalhador, args_lista)

print(f"Resultados: {resultados}")

Cada processo tem seu próprio PID (identificador único). O Pool gerencia automaticamente quantos processos rodam simultaneamente. Isso é mais eficiente que criar centenas de processos manuais.

Comunicação entre processos

Processos não compartilham memória como threads. Portanto, precisamos de mecanismos especiais para comunicação. O Queue permite trocar dados entre processos de forma segura. O Pipe oferece comunicação bidirecional entre dois processos. Já o Value e Array compartilham memória com locks. Quando usar cada um? Queue para produtor-consumidor. Pipe para comunicação simples entre dois processos. A voz passiva é aplicada: “os dados são serializados com pickle automaticamente”. Exemplo de comunicação com Queue:

import multiprocessing
import time

def produtor(queue, itens):
    """Produz itens e coloca na fila."""
    for item in itens:
        print(f"Produzindo: {item}")
        queue.put(item)
        time.sleep(0.5)
    queue.put(None)  # Sinal de fim

def consumidor(queue, nome):
    """Consome itens da fila."""
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"{nome} consumiu: {item}")
        time.sleep(0.3)

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    
    itens = ["Item1", "Item2", "Item3", "Item4", "Item5"]
    
    p_produtor = multiprocessing.Process(target=produtor, args=(queue, itens))
    p_consumidor1 = multiprocessing.Process(target=consumidor, args=(queue, "Consumidor-1"))
    p_consumidor2 = multiprocessing.Process(target=consumidor, args=(queue, "Consumidor-2"))
    
    p_produtor.start()
    p_consumidor1.start()
    p_consumidor2.start()
    
    p_produtor.join()
    p_consumidor1.join()
    p_consumidor2.join()
    
    # Exemplo com Value (memória compartilhada)
    print("\n=== Memória Compartilhada ===")
    contador = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock()
    
    def incrementar():
        with lock:
            contador.value += 1
    
    processos = []
    for _ in range(10):
        p = multiprocessing.Process(target=incrementar)
        processos.append(p)
        p.start()
    
    for p in processos:
        p.join()
    
    print(f"Valor final do contador: {contador.value}")

import multiprocessing

import time

def produtor(queue, itens):

"""Produz itens e coloca na fila."""

for item in itens:

print(f"Produzindo: {item}")

queue.put(item)

time.sleep(0.5)

queue.put(None) # Sinal de fim

def consumidor(queue, nome):

"""Consome itens da fila."""

while True:

item = queue.get()

if item is None:

break

print(f"{nome} consumiu: {item}")

time.sleep(0.3)

if __name__ == "__main__":

queue = multiprocessing.Queue()

itens = ["Item1", "Item2", "Item3", "Item4", "Item5"]

p_produtor = multiprocessing.Process(target=produtor, args=(queue, itens))

p_consumidor1 = multiprocessing.Process(target=consumidor, args=(queue, "Consumidor-1"))

p_consumidor2 = multiprocessing.Process(target=consumidor, args=(queue, "Consumidor-2"))

p_produtor.start()

p_consumidor1.start()

p_consumidor2.start()

p_produtor.join()

p_consumidor1.join()

p_consumidor2.join()

# Exemplo com Value (memória compartilhada)

print("\n=== Memória Compartilhada ===")

contador = multiprocessing.Value('i', 0)

lock = multiprocessing.Lock()

def incrementar():

with lock:

contador.value += 1

processos = []

for _ in range(10):

p = multiprocessing.Process(target=incrementar)

processos.append(p)

p.start()

for p in processos:

p.join()

print(f"Valor final do contador: {contador.value}")

Queues são ideais para padrões produtor-consumidor. Valores compartilhados exigem locks para evitar condições de corrida. A comunicação entre processos tem overhead, então use apenas quando necessário.

Pool e map para paralelismo de dados

A função pool.map() é a maneira mais fácil de paralelizar. Ela divide uma lista de dados entre os processos disponíveis. Cada processo aplica a mesma função a um subconjunto dos dados. Quando usar map? Em problemas de processamento de listas grandes. Por exemplo, aplicar uma função a cada elemento de 1 milhão de itens. A voz passiva é aplicada: “os resultados são coletados automaticamente”. Exemplo prático com processamento paralelo de números:

import multiprocessing
import time
import math

def processar_numero(n):
    """Função CPU-intensiva para testar paralelismo."""
    # Calcula muitos primos para simular trabalho pesado
    primos = []
    for i in range(2, min(10000, n)):
        eh_primo = True
        for j in range(2, int(math.sqrt(i)) + 1):
            if i % j == 0:
                eh_primo = False
                break
        if eh_primo:
            primos.append(i)
    return len(primos)

if __name__ == "__main__":
    numeros = list(range(10000, 10100))  # 100 números
    
    # Versão sequencial
    inicio_seq = time.time()
    resultados_seq = [processar_numero(n) for n in numeros]
    tempo_seq = time.time() - inicio_seq
    print(f"Sequencial: {tempo_seq:.2f}s")
    
    # Versão com multiprocessing
    inicio_par = time.time()
    with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:
        resultados_par = pool.map(processar_numero, numeros)
    tempo_par = time.time() - inicio_par
    
    print(f"Paralelo (com {multiprocessing.cpu_count()} CPUs): {tempo_par:.2f}s")
    print(f"Aceleração: {tempo_seq / tempo_par:.2f}x")
    
    # Verificando se os resultados são iguais
    assert resultados_seq == resultados_par
    print("Resultados idênticos!")
    
    # Usando map com múltiplos argumentos (starmap)
    def potencia(base, expoente):
        return base ** expoente
    
    argumentos = [(2, 10), (3, 8), (5, 5), (7, 4)]
    with multiprocessing.Pool() as pool:
        resultados_pot = pool.starmap(potencia, argumentos)
    print(f"Potências: {resultados_pot}")
    
    # Usando map_async para não-bloqueante
    with multiprocessing.Pool() as pool:
        resultado_async = pool.map_async(processar_numero, numeros[:10])
        print("Fazendo outra coisa enquanto processa...")
        resultados_async = resultado_async.get(timeout=30)
        print(f"Resultados async: {resultados_async[:3]}...")

import multiprocessing

import time

import math

def processar_numero(n):

"""Função CPU-intensiva para testar paralelismo."""

# Calcula muitos primos para simular trabalho pesado

primos = []

for i in range(2, min(10000, n)):

eh_primo = True

for j in range(2, int(math.sqrt(i)) + 1):

if i % j == 0:

eh_primo = False

break

if eh_primo:

primos.append(i)

return len(primos)

if __name__ == "__main__":

numeros = list(range(10000, 10100)) # 100 números

# Versão sequencial

inicio_seq = time.time()

resultados_seq = [processar_numero(n) for n in numeros]

tempo_seq = time.time() - inicio_seq

print(f"Sequencial: {tempo_seq:.2f}s")

# Versão com multiprocessing

inicio_par = time.time()

with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:

resultados_par = pool.map(processar_numero, numeros)

tempo_par = time.time() - inicio_par

print(f"Paralelo (com {multiprocessing.cpu_count()} CPUs): {tempo_par:.2f}s")

print(f"Aceleração: {tempo_seq / tempo_par:.2f}x")

# Verificando se os resultados são iguais

assert resultados_seq == resultados_par

print("Resultados idênticos!")

# Usando map com múltiplos argumentos (starmap)

def potencia(base, expoente):

return base ** expoente

argumentos = [(2, 10), (3, 8), (5, 5), (7, 4)]

with multiprocessing.Pool() as pool:

resultados_pot = pool.starmap(potencia, argumentos)

print(f"Potências: {resultados_pot}")

# Usando map_async para não-bloqueante

with multiprocessing.Pool() as pool:

resultado_async = pool.map_async(processar_numero, numeros[:10])

print("Fazendo outra coisa enquanto processa...")

resultados_async = resultado_async.get(timeout=30)

print(f"Resultados async: {resultados_async[:3]}...")

A aceleração deve ser próxima ao número de núcleos da CPU. Para 4 núcleos, espere cerca de 3.5x de ganho. A fórmula teórica é a lei de Amdahl: \(S = \frac{1}{(1 – P) + \frac{P}{N}}\) Onde P é a fração paralelizável e N é o número de núcleos. Multiprocessamento é a ferramenta certa para CPU-bound. Combine com boas práticas e evite overhead desnecessário. Seu código rodará muito mais rápido em máquinas modernas.

Multithreading em Python

22/05/202622/04/2026 Por antonino

Multithreading permite múltiplas threads dentro de um único processo. Cada thread executa um fluxo independente de instruções. Primeiramente, threads compartilham a mesma memória do processo. Por exemplo, duas threads podem acessar a mesma variável global. Isso facilita a comunicação entre elas. Além disso, threads são mais leves que processos completos. A voz passiva é usada aqui: “as threads são gerenciadas pelo sistema operacional”. Quando utilizar multithreading em Python? Em operações de I/O. Por exemplo, downloads de rede, leitura de arquivos ou consultas a banco. No entanto, o GIL (Global Interpreter Lock) limita a execução paralela. Portanto, threads não aceleram código com CPU intensivo. Vamos explorar criação, sincronização e boas práticas. Três subtítulos guiarão você pelo mundo do threading. Ao final, você dominará a concorrência em Python.

Criando e gerenciando threads

Python oferece o módulo threading para trabalhar com threads. A classe Thread representa uma thread executável. Para criar uma thread, instancie Thread(target=funcao). Depois, chame start() para iniciar a execução. O método join() espera a thread terminar. Quando usar criação manual? Em scripts com poucas threads controladas. A voz passiva é aplicada: “os argumentos são passados via args“. Exemplo básico de criação de threads:

import threading
import time

def trabalhador(nome, segundos):
    """Função executada por cada thread."""
    print(f"Thread {nome}: iniciando")
    time.sleep(segundos)
    print(f"Thread {nome}: finalizada após {segundos}s")

# Criando threads manualmente
t1 = threading.Thread(target=trabalhador, args=("A", 2))
t2 = threading.Thread(target=trabalhador, args=("B", 1))

# Iniciando as threads
t1.start()
t2.start()

# Aguardando ambas terminarem
t1.join()
t2.join()

print("Todas as threads finalizaram")

# Usando lista de threads
threads = []
for i in range(5):
    t = threading.Thread(target=trabalhador, args=(f"Thread-{i}", i % 3 + 1))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

import threading

import time

def trabalhador(nome, segundos):

"""Função executada por cada thread."""

print(f"Thread {nome}: iniciando")

time.sleep(segundos)

print(f"Thread {nome}: finalizada após {segundos}s")

# Criando threads manualmente

t1 = threading.Thread(target=trabalhador, args=("A", 2))

t2 = threading.Thread(target=trabalhador, args=("B", 1))

# Iniciando as threads

t1.start()

t2.start()

# Aguardando ambas terminarem

t1.join()

t2.join()

print("Todas as threads finalizaram")

# Usando lista de threads

threads = []

for i in range(5):

t = threading.Thread(target=trabalhador, args=(f"Thread-{i}", i % 3 + 1))

threads.append(t)

t.start()

for t in threads:

t.join()

Threads executam concorrentemente, não em paralelo real. O GIL alterna entre elas rapidamente, dando a ilusão de simultaneidade. Para I/O, isso é suficiente e muito eficiente.

Sincronização entre threads

Quando threads compartilham dados, podem ocorrer condições de corrida. Por exemplo, duas threads incrementando a mesma variável simultaneamente. Isso corrompe o resultado final. Para evitar isso, use mecanismos de sincronização. O Lock (mutex) permite que apenas uma thread execute uma seção crítica. Outros mecanismos incluem RLock, Semaphore e Event. Quando usar locks? Sempre que múltiplas threads acessarem dados compartilhados. A voz passiva é usada aqui: “os recursos compartilhados são protegidos por locks”. Exemplo de contador seguro com lock:

import threading
import time

# Sem sincronização (INCORRETO)
contador_sem_lock = 0

def incrementar_sem_lock():
    global contador_sem_lock
    for _ in range(100000):
        contador_sem_lock += 1

# Com sincronização (CORRETO)
contador_com_lock = 0
lock = threading.Lock()

def incrementar_com_lock():
    global contador_com_lock
    for _ in range(100000):
        with lock:  # Entrada na seção crítica
            contador_com_lock += 1
        # Saída automática da seção crítica

# Executando sem lock
threads_sem = []
for _ in range(5):
    t = threading.Thread(target=incrementar_sem_lock)
    threads_sem.append(t)
    t.start()

for t in threads_sem:
    t.join()
print(f"Sem lock: {contador_sem_lock} (esperado: 500000)")

# Executando com lock
threads_com = []
for _ in range(5):
    t = threading.Thread(target=incrementar_com_lock)
    threads_com.append(t)
    t.start()

for t in threads_com:
    t.join()
print(f"Com lock: {contador_com_lock} (esperado: 500000)")

# Exemplo de deadlock (evite!)
lock_a = threading.Lock()
lock_b = threading.Lock()

def tarefa_1():
    with lock_a:
        time.sleep(0.01)
        with lock_b:  # Pode causar deadlock com tarefa_2
            pass

def tarefa_2():
    with lock_b:
        time.sleep(0.01)
        with lock_a:  # Pode causar deadlock com tarefa_1
            pass

import threading

import time

# Sem sincronização (INCORRETO)

contador_sem_lock = 0

def incrementar_sem_lock():

global contador_sem_lock

for _ in range(100000):

contador_sem_lock += 1

# Com sincronização (CORRETO)

contador_com_lock = 0

lock = threading.Lock()

def incrementar_com_lock():

global contador_com_lock

for _ in range(100000):

with lock: # Entrada na seção crítica

contador_com_lock += 1

# Saída automática da seção crítica

# Executando sem lock

threads_sem = []

for _ in range(5):

t = threading.Thread(target=incrementar_sem_lock)

threads_sem.append(t)

t.start()

for t in threads_sem:

t.join()

print(f"Sem lock: {contador_sem_lock} (esperado: 500000)")

# Executando com lock

threads_com = []

for _ in range(5):

t = threading.Thread(target=incrementar_com_lock)

threads_com.append(t)

t.start()

for t in threads_com:

t.join()

print(f"Com lock: {contador_com_lock} (esperado: 500000)")

# Exemplo de deadlock (evite!)

lock_a = threading.Lock()

lock_b = threading.Lock()

def tarefa_1():

with lock_a:

time.sleep(0.01)

with lock_b: # Pode causar deadlock com tarefa_2

pass

def tarefa_2():

with lock_b:

time.sleep(0.01)

with lock_a: # Pode causar deadlock com tarefa_1

pass

O resultado sem lock será imprevisível e geralmente menor que 500000. Com lock, o resultado é sempre correto, porém mais lento. A fórmula do tempo de execução com contenção é: \(T = T_0 + C \times L\) Onde C é o número de aquisições e L é a latência do lock.

Threads vs. outras formas de concorrência

Threads são ótimas para I/O-bound com poucas conexões. Para CPU-bound, prefira multiprocessing (processos separados). Para milhares de conexões simultâneas, asyncio é superior. Quando escolher threading especificamente? Em projetos simples. Também quando você já tem código síncrono e não quer reescrever. Threads são mais fáceis de entender que asyncio. A desvantagem é o GIL e a complexidade de locks. A voz passiva é aplicada: “decisões de arquitetura são baseadas no tipo de tarefa”. Exemplo prático de pool de threads para tarefas I/O:

import threading
from concurrent.futures import ThreadPoolExecutor
import requests
import time

urls = [
    'https://httpbin.org/delay/1',
    'https://httpbin.org/delay/2',
    'https://httpbin.org/delay/0.5',
] * 5  # 15 URLs no total

def baixar(url):
    inicio = time.time()
    resposta = requests.get(url)
    return url, resposta.status_code, time.time() - inicio

# Usando ThreadPoolExecutor (mais elegante)
print("Usando ThreadPoolExecutor:")
with ThreadPoolExecutor(max_workers=5) as executor:
    resultados = executor.map(baixar, urls)
    for url, status, tempo in resultados:
        print(f"{url} -> status {status} em {tempo:.2f}s")

# Alternativa manual com lista de threads
print("\nUsando threads manuais:")
threads = []
resultados_manuais = []

def baixar_e_armazenar(url, idx):
    _, status, tempo = baixar(url)
    resultados_manuais.append((idx, status, tempo))

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

for t in threads:
    t.join()

import threading

from concurrent.futures import ThreadPoolExecutor

import requests

import time

urls = [

'https://httpbin.org/delay/1',

'https://httpbin.org/delay/2',

'https://httpbin.org/delay/0.5',

] * 5 # 15 URLs no total

def baixar(url):

inicio = time.time()

resposta = requests.get(url)

return url, resposta.status_code, time.time() - inicio

# Usando ThreadPoolExecutor (mais elegante)

print("Usando ThreadPoolExecutor:")

with ThreadPoolExecutor(max_workers=5) as executor:

resultados = executor.map(baixar, urls)

for url, status, tempo in resultados:

print(f"{url} -> status {status} em {tempo:.2f}s")

# Alternativa manual com lista de threads

print("\nUsando threads manuais:")

threads = []

resultados_manuais = []

def baixar_e_armazenar(url, idx):

_, status, tempo = baixar(url)

resultados_manuais.append((idx, status, tempo))

for i, url in enumerate(urls):

t = threading.Thread(target=baixar_e_armazenar, args=(url, i))

threads.append(t)

t.start()

for t in threads:

t.join()

ThreadPoolExecutor simplifica o gerenciamento de threads. Ele reutiliza threads para evitar custo de criação. Use max_workers entre 5 e 10 para tarefas de rede. Nunca crie mais de 100 threads simultâneas. O sistema operacional pode ficar sobrecarregado. Threads são ferramentas poderosas quando usadas corretamente. Domine os locks e evite deadlocks. Sua aplicação ficará mais rápida e responsiva. Comece com pequenos exemplos e escale gradualmente.