antonino, Autor em Área de Trampo

Processos Paralelos em Python: Paralelismo Verdadeiro

22/05/202622/04/2026 Por antonino

Processos paralelos executam código simultaneamente em múltiplos núcleos. Isso é chamado de true parallelism ou paralelismo verdadeiro. Primeiramente, cada processo tem seu próprio interpretador e memória. Por exemplo, 8 processos podem rodar em 8 núcleos de CPU ao mesmo tempo. Além disso, processos não compartilham o GIL (Global Interpreter Lock). A voz passiva é usada aqui: “o sistema operacional gerencia a distribuição entre núcleos”. Quando utilizar processos paralelos? Em tarefas com CPU intensiva. Por exemplo, processamento de imagens, simulações científicas ou machine learning. Também em qualquer tarefa que exija todo o poder da máquina. Python oferece o módulo multiprocessing e o concurrent.futures. Vamos explorar características, ganhos e limitações. Três subtítulos guiarão você pelo verdadeiro paralelismo. Ao final, você dominará a execução paralela em Python.

Multiprocessing vs. threading: a diferença crucial

Threads são leves, mas limitadas pelo GIL em CPU-bound. Processos são mais pesados, mas executam em paralelo verdadeiro. Quando escolher processos? Quando o gargalo é CPU, não I/O. A criação de um processo é mais cara que uma thread. No entanto, o ganho em paralelismo compensa para tarefas longas. A voz passiva é aplicada: “a memória é duplicada para cada processo”. Exemplo comparativo entre threads e processos para CPU-bound:

import multiprocessing
import threading
import time
import os

def trabalho_cpu_pesado(segundos):
    """Função que consome muita CPU."""
    pid = os.getpid()
    inicio = time.time()
    # Loop pesado para simular trabalho
    total = 0
    for i in range(30_000_000):
        total += i * i
    fim = time.time()
    print(f"Processo/Thread PID {pid} concluiu em {fim - inicio:.2f}s")
    return total

def executar_com_processos(n_tarefas):
    """Executa com processos paralelos."""
    with multiprocessing.Pool(processes=n_tarefas) as pool:
        resultados = pool.map(trabalho_cpu_pesado, [1] * n_tarefas)
    return resultados

def executar_com_threads(n_tarefas):
    """Executa com threads (limitado pelo GIL)."""
    threads = []
    resultados = [0] * n_tarefas
    
    def wrapper(idx):
        resultados[idx] = trabalho_cpu_pesado(1)
    
    for i in range(n_tarefas):
        t = threading.Thread(target=wrapper, args=(i,))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    return resultados

if __name__ == "__main__":
    n = 4  # Número de tarefas (igual aos núcleos)
    
    print(f"Executando {n} tarefas CPU-bound com PROCESSOS:")
    inicio = time.time()
    executar_com_processos(n)
    tempo_processos = time.time() - inicio
    print(f"Tempo total com processos: {tempo_processos:.2f}s")
    
    print(f"\nExecutando {n} tarefas CPU-bound com THREADS:")
    inicio = time.time()
    executar_com_threads(n)
    tempo_threads = time.time() - inicio
    print(f"Tempo total com threads: {tempo_threads:.2f}s")
    
    print(f"\nProcessos foram {tempo_threads / tempo_processos:.1f}x mais rápidos!")
    print("(Processos executam em paralelo verdadeiro, threads não)")

import multiprocessing

import threading

import time

import os

def trabalho_cpu_pesado(segundos):

"""Função que consome muita CPU."""

pid = os.getpid()

inicio = time.time()

# Loop pesado para simular trabalho

total = 0

for i in range(30_000_000):

total += i * i

fim = time.time()

print(f"Processo/Thread PID {pid} concluiu em {fim - inicio:.2f}s")

return total

def executar_com_processos(n_tarefas):

"""Executa com processos paralelos."""

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

resultados = pool.map(trabalho_cpu_pesado, [1] * n_tarefas)

return resultados

def executar_com_threads(n_tarefas):

"""Executa com threads (limitado pelo GIL)."""

threads = []

resultados = [0] * n_tarefas

def wrapper(idx):

resultados[idx] = trabalho_cpu_pesado(1)

for i in range(n_tarefas):

t = threading.Thread(target=wrapper, args=(i,))

threads.append(t)

t.start()

for t in threads:

t.join()

return resultados

if __name__ == "__main__":

n = 4 # Número de tarefas (igual aos núcleos)

print(f"Executando {n} tarefas CPU-bound com PROCESSOS:")

inicio = time.time()

executar_com_processos(n)

tempo_processos = time.time() - inicio

print(f"Tempo total com processos: {tempo_processos:.2f}s")

print(f"\nExecutando {n} tarefas CPU-bound com THREADS:")

inicio = time.time()

executar_com_threads(n)

tempo_threads = time.time() - inicio

print(f"Tempo total com threads: {tempo_threads:.2f}s")

print(f"\nProcessos foram {tempo_threads / tempo_processos:.1f}x mais rápidos!")

print("(Processos executam em paralelo verdadeiro, threads não)")

Processos mostram aceleração próxima ao número de núcleos. Threads mostram aceleração próxima de 1.0 devido ao GIL. Essa é a diferença fundamental entre concorrência e paralelismo.

Comunicação e sincronização entre processos

Processos paralelos precisam se comunicar para trocar resultados. Diferente de threads, eles não compartilham memória automaticamente. Portanto, usamos filas (Queue), pipes ou memória compartilhada. A serialização (pickle) é necessária para enviar dados entre processos. Isso adiciona overhead, mas é inevitável para isolamento. Quando usar memória compartilhada? Para grandes arrays numéricos. A voz passiva é usada aqui: “os dados são copiados por valor entre processos”. Exemplo de comunicação com Queue entre processos paralelos:

import multiprocessing
import time
import random

def produtor(queue, id_produtor, n_itens):
    """Produz itens e coloca na fila."""
    for i in range(n_itens):
        item = f"Produtor-{id_produtor}-Item-{i}"
        queue.put(item)
        print(f"[Produtor {id_produtor}] Produziu: {item}")
        time.sleep(random.uniform(0.1, 0.3))
    queue.put(None)  # Sinal de fim para este produtor

def consumidor(queue, id_consumidor):
    """Consome itens da fila até receber Nones suficientes."""
    n_finalizados = 0
    while True:
        item = queue.get()
        if item is None:
            n_finalizados += 1
            if n_finalizados == 2:  # 2 produtores
                break
            continue
        print(f"[Consumidor {id_consumidor}] Consumiu: {item}")
        # Simula processamento do item
        time.sleep(random.uniform(0.2, 0.5))

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    
    # Criando produtores paralelos
    produtores = []
    for i in range(2):
        p = multiprocessing.Process(target=produtor, args=(queue, i, 5))
        produtores.append(p)
        p.start()
    
    # Criando consumidores paralelos
    consumidores = []
    for i in range(2):
        c = multiprocessing.Process(target=consumidor, args=(queue, i))
        consumidores.append(c)
        c.start()
    
    # Aguardar todos terminarem
    for p in produtores:
        p.join()
    for c in consumidores:
        c.join()
    
    print("Todos os processos finalizaram")
    
    # Exemplo com memória compartilhada (Array)
    print("\n=== Memória Compartilhada com Array ===")
    from multiprocessing import Array, Lock
    
    arr = Array('i', [0] * 10)  # Array de 10 inteiros
    lock = Lock()
    
    def atualizar_array(indice, valor):
        with lock:
            arr[indice] = valor
    
    processos = []
    for i in range(10):
        p = multiprocessing.Process(target=atualizar_array, args=(i, i * 10))
        processos.append(p)
        p.start()
    
    for p in processos:
        p.join()
    
    print(f"Array final: {list(arr)}")

import multiprocessing

import time

import random

def produtor(queue, id_produtor, n_itens):

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

for i in range(n_itens):

item = f"Produtor-{id_produtor}-Item-{i}"

queue.put(item)

print(f"[Produtor {id_produtor}] Produziu: {item}")

time.sleep(random.uniform(0.1, 0.3))

queue.put(None) # Sinal de fim para este produtor

def consumidor(queue, id_consumidor):

"""Consome itens da fila até receber Nones suficientes."""

n_finalizados = 0

while True:

item = queue.get()

if item is None:

n_finalizados += 1

if n_finalizados == 2: # 2 produtores

break

continue

print(f"[Consumidor {id_consumidor}] Consumiu: {item}")

# Simula processamento do item

time.sleep(random.uniform(0.2, 0.5))

if __name__ == "__main__":

queue = multiprocessing.Queue()

# Criando produtores paralelos

produtores = []

for i in range(2):

p = multiprocessing.Process(target=produtor, args=(queue, i, 5))

produtores.append(p)

p.start()

# Criando consumidores paralelos

consumidores = []

for i in range(2):

c = multiprocessing.Process(target=consumidor, args=(queue, i))

consumidores.append(c)

c.start()

# Aguardar todos terminarem

for p in produtores:

p.join()

for c in consumidores:

c.join()

print("Todos os processos finalizaram")

# Exemplo com memória compartilhada (Array)

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

from multiprocessing import Array, Lock

arr = Array('i', [0] * 10) # Array de 10 inteiros

lock = Lock()

def atualizar_array(indice, valor):

with lock:

arr[indice] = valor

processos = []

for i in range(10):

p = multiprocessing.Process(target=atualizar_array, args=(i, i * 10))

processos.append(p)

p.start()

for p in processos:

p.join()

print(f"Array final: {list(arr)}")

Queues são seguras e ideais para padrões produtor-consumidor. Arrays compartilhados são eficientes para dados numéricos. Sempre use locks ao modificar memória compartilhada.

Escalabilidade e lei de amdahl

O ganho com processos paralelos não é linear infinito. A lei de Amdahl descreve o limite teórico de aceleração. A fórmula é: \(S = \frac{1}{(1 – P) + \frac{P}{N}}\) Onde P é a fração paralelizável e N é o número de núcleos. Por exemplo, com 90% paralelizável e 8 núcleos, o ganho máximo é 4.7x. Portanto, nem todo código pode ser perfeitamente paralelizado. A voz passiva é aplicada: “partes sequenciais são executadas em apenas um núcleo”. Exemplo demonstrando a lei de Amdahl na prática:

import multiprocessing
import time

def trabalho_paralelizavel(n):
    """Trabalho que pode ser dividido entre processos."""
    total = 0
    for i in range(n):
        total += i * i
    return total

def trabalho_sequencial(n):
    """Trabalho que precisa rodar em um único núcleo."""
    total = 0
    for i in range(n):
        total += i ** 0.5
    return total

def executar_com_paralelismo(n_tarefas, tamanho_par, tamanho_seq):
    """Executa parte paralela + parte sequencial."""
    # Parte sequencial (um núcleo)
    resultado_seq = trabalho_sequencial(tamanho_seq)
    
    # Parte paralela (múltiplos núcleos)
    with multiprocessing.Pool(processes=n_tarefas) as pool:
        resultados_par = pool.map(trabalho_paralelizavel, [tamanho_par] * n_tarefas)
    
    return resultado_seq, sum(resultados_par)

if __name__ == "__main__":
    n_nucleos = multiprocessing.cpu_count()
    print(f"Máquina com {n_nucleos} núcleos\n")
    
    # Experimento 1: 90% paralelizável
    tamanho_par = 40_000_000
    tamanho_seq = int(tamanho_par * 0.111)  # ~10% sequencial
    
    inicio = time.time()
    executar_com_paralelismo(1, tamanho_par, tamanho_seq)
    tempo_1_nucleo = time.time() - inicio
    
    inicio = time.time()
    executar_com_paralelismo(n_nucleos, tamanho_par, tamanho_seq)
    tempo_n_nucleos = time.time() - inicio
    
    aceleracao = tempo_1_nucleo / tempo_n_nucleos
    print(f"90% paralelizável:")
    print(f"  1 núcleo: {tempo_1_nucleo:.2f}s")
    print(f"  {n_nucleos} núcleos: {tempo_n_nucleos:.2f}s")
    print(f"  Aceleração real: {aceleracao:.2f}x")
    
    # Experimento 2: 50% paralelizável
    tamanho_par = 20_000_000
    tamanho_seq = tamanho_par  # 50% cada
    
    inicio = time.time()
    executar_com_paralelismo(1, tamanho_par, tamanho_seq)
    tempo_1_nucleo = time.time() - inicio
    
    inicio = time.time()
    executar_com_paralelismo(n_nucleos, tamanho_par, tamanho_seq)
    tempo_n_nucleos = time.time() - inicio
    
    aceleracao = tempo_1_nucleo / tempo_n_nucleos
    print(f"\n50% paralelizável:")
    print(f"  1 núcleo: {tempo_1_nucleo:.2f}s")
    print(f"  {n_nucleos} núcleos: {tempo_n_nucleos:.2f}s")
    print(f"  Aceleração real: {aceleracao:.2f}x")

import multiprocessing

import time

def trabalho_paralelizavel(n):

"""Trabalho que pode ser dividido entre processos."""

total = 0

for i in range(n):

total += i * i

return total

def trabalho_sequencial(n):

"""Trabalho que precisa rodar em um único núcleo."""

total = 0

for i in range(n):

total += i ** 0.5

return total

def executar_com_paralelismo(n_tarefas, tamanho_par, tamanho_seq):

"""Executa parte paralela + parte sequencial."""

# Parte sequencial (um núcleo)

resultado_seq = trabalho_sequencial(tamanho_seq)

# Parte paralela (múltiplos núcleos)

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

resultados_par = pool.map(trabalho_paralelizavel, [tamanho_par] * n_tarefas)

return resultado_seq, sum(resultados_par)

if __name__ == "__main__":

n_nucleos = multiprocessing.cpu_count()

print(f"Máquina com {n_nucleos} núcleos\n")

# Experimento 1: 90% paralelizável

tamanho_par = 40_000_000

tamanho_seq = int(tamanho_par * 0.111) # ~10% sequencial

inicio = time.time()

executar_com_paralelismo(1, tamanho_par, tamanho_seq)

tempo_1_nucleo = time.time() - inicio

inicio = time.time()

executar_com_paralelismo(n_nucleos, tamanho_par, tamanho_seq)

tempo_n_nucleos = time.time() - inicio

aceleracao = tempo_1_nucleo / tempo_n_nucleos

print(f"90% paralelizável:")

print(f" 1 núcleo: {tempo_1_nucleo:.2f}s")

print(f" {n_nucleos} núcleos: {tempo_n_nucleos:.2f}s")

print(f" Aceleração real: {aceleracao:.2f}x")

# Experimento 2: 50% paralelizável

tamanho_par = 20_000_000

tamanho_seq = tamanho_par # 50% cada

inicio = time.time()

executar_com_paralelismo(1, tamanho_par, tamanho_seq)

tempo_1_nucleo = time.time() - inicio

inicio = time.time()

executar_com_paralelismo(n_nucleos, tamanho_par, tamanho_seq)

tempo_n_nucleos = time.time() - inicio

aceleracao = tempo_1_nucleo / tempo_n_nucleos

print(f"\n50% paralelizável:")

print(f" 1 núcleo: {tempo_1_nucleo:.2f}s")

print(f" {n_nucleos} núcleos: {tempo_n_nucleos:.2f}s")

print(f" Aceleração real: {aceleracao:.2f}x")

Observe que quanto maior a fração sequencial, menor o ganho. Portanto, identifique e otimize os gargalos sequenciais primeiro. Processos paralelos são ferramentas poderosas, mas não mágicas. Use-os com sabedoria e meça sempre o ganho real. Para muitos problemas, o paralelismo verdadeiro transforma horas em minutos. Experimente e veja seu código voar em múltiplos núcleos.

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.