Arquivo de Distribuída - Área de Trampo

Ray Computação Distribuída

22/05/202622/04/2026 Por antonino

Ray é um framework para computação distribuída em Python. Ele transforma funções e classes em tarefas remotas facilmente. Primeiramente, Ray escala de um laptop para um cluster gigante. Por exemplo, você pode paralelizar loops sem modificar muito código. Além disso, Ray é excelente para machine learning e reinforcement learning. Assim, cientistas de dados ganham produtividade imensa. Consequentemente, projetos que levavam dias agora levam horas. Portanto, a adoção de Ray cresce rapidamente na indústria. Quando utilizar Ray? Em computação paralela de larga escala. Também em treinamento de modelos, otimização de hiperparâmetros e simulações. Da mesma forma, qualquer tarefa paralelizável se beneficia. Por outro lado, para tarefas sequenciais simples, Ray é desnecessário. Ray oferece atores (estado compartilhado) e tarefas sem estado. Então, vamos explorar instalação, conceitos e exemplos práticos. Três subtítulos guiarão você pelo universo do Ray. Portanto, ao final, você paralelizará código como nunca antes.

Instalação e primeiros passos com ray

Instale Ray com pip: pip install ray. Depois, importe ray e inicie com ray.init(). O decorador @ray.remote torna uma função distribuída. Chamá-la com .remote() cria uma tarefa assíncrona. Use ray.get() para obter os resultados. Quando usar tarefas Ray? Em operações independentes e paralelizáveis. Por exemplo, processar imagens ou calcular estatísticas. Além disso, tarefas Ray são ideais para map-reduce simples. Exemplo básico de Ray:

import ray
import time

# Inicializa o Ray (local, sem cluster)
ray.init(ignore_reinit_error=True)

@ray.remote
def tarefa_pesada(n):
    """Função que será executada remotamente."""
    time.sleep(1)  # Simula trabalho pesado
    return n * n

# Versão sequencial
print("=== Execução sequencial ===")
inicio_seq = time.time()
resultados_seq = [tarefa_pesada(i) for i in range(10)]
tempo_seq = time.time() - inicio_seq
print(f"Tempo sequencial: {tempo_seq:.2f}s")

# Versão com Ray (paralela)
print("\n=== Execução com Ray ===")
inicio_ray = time.time()
# Cria tarefas remotas (não bloqueia)
futures = [tarefa_pesada.remote(i) for i in range(10)]
# Aguarda os resultados
resultados_ray = ray.get(futures)
tempo_ray = time.time() - inicio_ray
print(f"Tempo com Ray: {tempo_ray:.2f}s")
print(f"Aceleração: {tempo_seq / tempo_ray:.2f}x")

# Verificando resultados
print(f"Resultados: {resultados_ray[:5]}...")

# Fecha o Ray
ray.shutdown()

import ray

import time

# Inicializa o Ray (local, sem cluster)

ray.init(ignore_reinit_error=True)

@ray.remote

def tarefa_pesada(n):

"""Função que será executada remotamente."""

time.sleep(1) # Simula trabalho pesado

return n * n

# Versão sequencial

print("=== Execução sequencial ===")

inicio_seq = time.time()

resultados_seq = [tarefa_pesada(i) for i in range(10)]

tempo_seq = time.time() - inicio_seq

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

# Versão com Ray (paralela)

print("\n=== Execução com Ray ===")

inicio_ray = time.time()

# Cria tarefas remotas (não bloqueia)

futures = [tarefa_pesada.remote(i) for i in range(10)]

# Aguarda os resultados

resultados_ray = ray.get(futures)

tempo_ray = time.time() - inicio_ray

print(f"Tempo com Ray: {tempo_ray:.2f}s")

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

# Verificando resultados

print(f"Resultados: {resultados_ray[:5]}...")

# Fecha o Ray

ray.shutdown()

Ray executa tarefas em paralelo usando todos os núcleos. O ganho é próximo ao número de CPUs disponíveis. Portanto, para tarefas independentes, Ray é imbatível.

Atores: estado compartilhado entre tarefas

Atores são objetos Ray que mantêm estado entre chamadas. Use @ray.remote em uma classe para criar um ator. Instancie com Classe.remote() e chame métodos com .metodo.remote(). Quando usar atores? Em cenários com estado compartilhado. Por exemplo, contadores, acumuladores ou cache distribuído. Além disso, atores são úteis para simulações ou parâmetros compartilhados. Assim, você mantém consistência sem locks complexos. Atores mantêm estado consistente mesmo com chamadas concorrentes. Então, para sistemas com estado, prefira atores a tarefas puras. Exemplo de ator contador distribuído:

import ray
import time

ray.init(ignore_reinit_error=True)

@ray.remote
class ContadorDistribuido:
    """Ator que mantém um contador compartilhado."""
    
    def __init__(self):
        self.valor = 0
    
    def incrementar(self, quantidade=1):
        self.valor += quantidade
        return self.valor
    
    def obter_valor(self):
        return self.valor
    
    def resetar(self):
        self.valor = 0
        return self.valor

# Criando um ator (instância remota)
contador = ContadorDistribuido.remote()

# Chamando métodos do ator remotamente
print("=== Ator Contador Distribuído ===")
futures = []
for i in range(10):
    # Incrementa de forma distribuída
    futures.append(contador.incrementar.remote(1))

# Aguarda resultados parciais
resultados = ray.get(futures)
print(f"Incrementos: {resultados}")

# Obtém valor final
valor_final = ray.get(contador.obter_valor.remote())
print(f"Valor final do contador: {valor_final}")

# Exemplo com múltiplos atores independentes
print("\n=== Múltiplos Atores ===")
atores = [ContadorDistribuido.remote() for _ in range(5)]

# Incrementa cada ator com diferentes valores
for i, ator in enumerate(atores):
    ator.incrementar.remote(i + 1)

# Coleta resultados
valores = ray.get([ator.obter_valor.remote() for ator in atores])
print(f"Valores dos atores: {valores}")

# Exemplo com ator que acumula dados
@ray.remote
class Acumulador:
    def __init__(self):
        self.dados = []
    
    def adicionar(self, item):
        self.dados.append(item)
        return len(self.dados)
    
    def obter_todos(self):
        return self.dados
    
    def media(self):
        if not self.dados:
            return 0
        return sum(self.dados) / len(self.dados)

acumulador = Acumulador.remote()
for i in range(20):
    acumulador.adicionar.remote(i * 2)

# Obtém resultados
tamanho = ray.get(acumulador.adicionar.remote(100))
media = ray.get(acumulador.media.remote())
todos = ray.get(acumulador.obter_todos.remote())

print(f"\n=== Acumulador ===")
print(f"Total de itens: {tamanho}")
print(f"Média: {media:.2f}")
print(f"Primeiros 5 itens: {todos[:5]}...")

ray.shutdown()

import ray

import time

ray.init(ignore_reinit_error=True)

@ray.remote

class ContadorDistribuido:

"""Ator que mantém um contador compartilhado."""

def __init__(self):

self.valor = 0

def incrementar(self, quantidade=1):

self.valor += quantidade

return self.valor

def obter_valor(self):

return self.valor

def resetar(self):

self.valor = 0

return self.valor

# Criando um ator (instância remota)

contador = ContadorDistribuido.remote()

# Chamando métodos do ator remotamente

print("=== Ator Contador Distribuído ===")

futures = []

for i in range(10):

# Incrementa de forma distribuída

futures.append(contador.incrementar.remote(1))

# Aguarda resultados parciais

resultados = ray.get(futures)

print(f"Incrementos: {resultados}")

# Obtém valor final

valor_final = ray.get(contador.obter_valor.remote())

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

# Exemplo com múltiplos atores independentes

print("\n=== Múltiplos Atores ===")

atores = [ContadorDistribuido.remote() for _ in range(5)]

# Incrementa cada ator com diferentes valores

for i, ator in enumerate(atores):

ator.incrementar.remote(i + 1)

# Coleta resultados

valores = ray.get([ator.obter_valor.remote() for ator in atores])

print(f"Valores dos atores: {valores}")

# Exemplo com ator que acumula dados

@ray.remote

class Acumulador:

def __init__(self):

self.dados = []

def adicionar(self, item):

self.dados.append(item)

return len(self.dados)

def obter_todos(self):

return self.dados

def media(self):

if not self.dados:

return 0

return sum(self.dados) / len(self.dados)

acumulador = Acumulador.remote()

for i in range(20):

acumulador.adicionar.remote(i * 2)

# Obtém resultados

tamanho = ray.get(acumulador.adicionar.remote(100))

media = ray.get(acumulador.media.remote())

todos = ray.get(acumulador.obter_todos.remote())

print(f"\n=== Acumulador ===")

print(f"Total de itens: {tamanho}")

print(f"Média: {media:.2f}")

print(f"Primeiros 5 itens: {todos[:5]}...")

ray.shutdown()

Atores mantêm estado entre chamadas remotas. Isso é fundamental para muitos algoritmos distribuídos. Assim, você implementa sistemas complexos com facilidade.

Ray para machine learning e tunning de hiperparâmetros

Ray tem bibliotecas especializadas como Ray Tune e Ray Train. Ray Tune otimiza hiperparâmetros de modelos de ML. Ray Train distribui treinamento de modelos grandes. Quando usar essas ferramentas? Em projetos de ML profissional. Também em otimização de modelos com muitos parâmetros. Por exemplo, redes neurais ou algoritmos de ensemble. Além disso, Ray Serve serve modelos em produção. Exemplo de Ray Tune para otimização:

import ray
from ray import tune
from ray.tune.schedulers import ASHAScheduler
import numpy as np
import time

# Inicializa Ray
ray.init(ignore_reinit_error=True)

def funcao_objetivo(config):
    """Função a ser otimizada (exemplo: SVM com parâmetros)."""
    # Simula treinamento de modelo
    x = config["x"]
    y = config["y"]
    z = config["z"]
    
    # Função de custo simulada
    custo = (x - 1)**2 + (y - 2)**2 + (z - 3)**2
    time.sleep(0.1)  # Simula tempo de treinamento
    
    # Reporta métricas para o Tune
    tune.report(loss=custo, accuracy=1.0 / (1.0 + custo))

def demonstrar_ray_tune():
    """Demonstra otimização de hiperparâmetros com Ray Tune."""
    print("=== Ray Tune - Otimização de Hiperparâmetros ===\n")
    
    # Espaço de busca
    espaco_busca = {
        "x": tune.uniform(0, 2),
        "y": tune.uniform(1, 3),
        "z": tune.uniform(2, 4),
    }
    
    # Executa otimização
    analise = tune.run(
        funcao_objetivo,
        config=espaco_busca,
        num_samples=20,  # Número de combinações
        metric="loss",
        mode="min",
        progress_bar=True
    )
    
    # Melhores parâmetros encontrados
    melhores_config = analise.get_best_config(metric="loss", mode="min")
    melhor_loss = analise.get_best_trial(metric="loss", mode="min").last_result["loss"]
    
    print(f"\nMelhores parâmetros: x={melhores_config['x']:.3f}, "
          f"y={melhores_config['y']:.3f}, z={melhores_config['z']:.3f}")
    print(f"Melhor loss: {melhor_loss:.6f}")
    
    return melhores_config

# Exemplo com Ray para processamento paralelo de dados
@ray.remote
def processar_lote(dados_lote):
    """Processa um lote de dados em paralelo."""
    resultado = [x * 2 for x in dados_lote]
    time.sleep(0.01)  # Simula processamento
    return sum(resultado)

def processamento_paralelo_dados():
    """Divide dados em lotes e processa em paralelo."""
    dados = list(range(1000))
    num_lotes = 8
    tamanho_lote = len(dados) // num_lotes
    
    lotes = []
    for i in range(num_lotes):
        inicio = i * tamanho_lote
        fim = inicio + tamanho_lote if i < num_lotes - 1 else len(dados)
        lotes.append(dados[inicio:fim])
    
    print("\n=== Processamento Paralelo de Dados ===")
    inicio = time.time()
    
    # Processa lotes em paralelo
    futures = [processar_lote.remote(lote) for lote in lotes]
    resultados = ray.get(futures)
    total = sum(resultados)
    
    tempo = time.time() - inicio
    print(f"Processados {len(dados)} itens em {tempo:.3f}s")
    print(f"Total: {total}")
    
    # Comparação sequencial
    inicio_seq = time.time()
    total_seq = sum(x * 2 for x in dados)
    tempo_seq = time.time() - inicio_seq
    print(f"Tempo sequencial: {tempo_seq:.3f}s")
    print(f"Aceleração: {tempo_seq / tempo:.2f}x")

# Exemplo com Ray para simulação distribuída
@ray.remote
def simulacao(parametros):
    """Simula um sistema com parâmetros dados."""
    # Simulação complexa
    resultado = np.sin(parametros[0]) * np.cos(parametros[1])
    return resultado

def simulacoes_distribuidas():
    """Executa múltiplas simulações em paralelo."""
    print("\n=== Simulações Distribuídas ===")
    parametros_lista = [(i * 0.1, j * 0.1) for i in range(10) for j in range(10)]
    
    inicio = time.time()
    futures = [simulacao.remote(p) for p in parametros_lista]
    resultados = ray.get(futures)
    tempo = time.time() - inicio
    
    print(f"Executadas {len(parametros_lista)} simulações em {tempo:.2f}s")
    print(f"Média dos resultados: {np.mean(resultados):.4f}")

if __name__ == "__main__":
    # Demonstrações
    melhores = demonstrar_ray_tune()
    processamento_paralelo_dados()
    simulacoes_distribuidas()
    
    print("\n=== Resumo do Ray ===")
    print("Ray é ideal para:")
    print("  - Paralelização de loops pesados")
    print("  - Atores com estado compartilhado")
    print("  - Otimização de hiperparâmetros")
    print("  - Treinamento distribuído de ML")
    
    ray.shutdown()

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

import ray

from ray import tune

from ray.tune.schedulers import ASHAScheduler

import numpy as np

import time

# Inicializa Ray

ray.init(ignore_reinit_error=True)

def funcao_objetivo(config):

"""Função a ser otimizada (exemplo: SVM com parâmetros)."""

# Simula treinamento de modelo

x = config["x"]

y = config["y"]

z = config["z"]

# Função de custo simulada

custo = (x - 1)**2 + (y - 2)**2 + (z - 3)**2

time.sleep(0.1) # Simula tempo de treinamento

# Reporta métricas para o Tune

tune.report(loss=custo, accuracy=1.0 / (1.0 + custo))

def demonstrar_ray_tune():

"""Demonstra otimização de hiperparâmetros com Ray Tune."""

print("=== Ray Tune - Otimização de Hiperparâmetros ===\n")

# Espaço de busca

espaco_busca = {

"x": tune.uniform(0, 2),

"y": tune.uniform(1, 3),

"z": tune.uniform(2, 4),

}

# Executa otimização

analise = tune.run(

funcao_objetivo,

config=espaco_busca,

num_samples=20, # Número de combinações

metric="loss",

mode="min",

progress_bar=True

)

# Melhores parâmetros encontrados

melhores_config = analise.get_best_config(metric="loss", mode="min")

melhor_loss = analise.get_best_trial(metric="loss", mode="min").last_result["loss"]

print(f"\nMelhores parâmetros: x={melhores_config['x']:.3f}, "

f"y={melhores_config['y']:.3f}, z={melhores_config['z']:.3f}")

print(f"Melhor loss: {melhor_loss:.6f}")

return melhores_config

# Exemplo com Ray para processamento paralelo de dados

@ray.remote

def processar_lote(dados_lote):

"""Processa um lote de dados em paralelo."""

resultado = [x * 2 for x in dados_lote]

time.sleep(0.01) # Simula processamento

return sum(resultado)

def processamento_paralelo_dados():

"""Divide dados em lotes e processa em paralelo."""

dados = list(range(1000))

num_lotes = 8

tamanho_lote = len(dados) // num_lotes

lotes = []

for i in range(num_lotes):

inicio = i * tamanho_lote

fim = inicio + tamanho_lote if i < num_lotes - 1 else len(dados)

lotes.append(dados[inicio:fim])

print("\n=== Processamento Paralelo de Dados ===")

inicio = time.time()

# Processa lotes em paralelo

futures = [processar_lote.remote(lote) for lote in lotes]

resultados = ray.get(futures)

total = sum(resultados)

tempo = time.time() - inicio

print(f"Processados {len(dados)} itens em {tempo:.3f}s")

print(f"Total: {total}")

# Comparação sequencial

inicio_seq = time.time()

total_seq = sum(x * 2 for x in dados)

tempo_seq = time.time() - inicio_seq

print(f"Tempo sequencial: {tempo_seq:.3f}s")

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

# Exemplo com Ray para simulação distribuída

@ray.remote

def simulacao(parametros):

"""Simula um sistema com parâmetros dados."""

# Simulação complexa

resultado = np.sin(parametros[0]) * np.cos(parametros[1])

return resultado

def simulacoes_distribuidas():

"""Executa múltiplas simulações em paralelo."""

print("\n=== Simulações Distribuídas ===")

parametros_lista = [(i * 0.1, j * 0.1) for i in range(10) for j in range(10)]

inicio = time.time()

futures = [simulacao.remote(p) for p in parametros_lista]

resultados = ray.get(futures)

tempo = time.time() - inicio

print(f"Executadas {len(parametros_lista)} simulações em {tempo:.2f}s")

print(f"Média dos resultados: {np.mean(resultados):.4f}")

if __name__ == "__main__":

# Demonstrações

melhores = demonstrar_ray_tune()

processamento_paralelo_dados()

simulacoes_distribuidas()

print("\n=== Resumo do Ray ===")

print("Ray é ideal para:")

print(" - Paralelização de loops pesados")

print(" - Atores com estado compartilhado")

print(" - Otimização de hiperparâmetros")

print(" - Treinamento distribuído de ML")

ray.shutdown()

Ray Tune busca automaticamente os melhores parâmetros. A fórmula do ganho em paralelismo é clara: \(S = \frac{T_{\text{seq}}}{T_{\text{ray}}} \approx N_{\text{workers}} \times (1 – O)\) Onde O é o overhead de comunicação (geralmente pequeno). Portanto, Ray é a ferramenta moderna para computação distribuída. Comece com tarefas simples e evolua para atores complexos. Finalmente, seu código escalará de um laptop para um cluster inteiro.

Programação Distribuída em Python

22/05/202622/04/2026 Por antonino

Programação distribuída executa código em vários computadores ao mesmo tempo. Ela conecta máquinas através de uma rede para resolver problemas grandes. Primeiramente, isso permite escalar horizontalmente além de um único servidor. Por exemplo, processar terabytes de dados com 100 máquinas em paralelo. Além disso, sistemas distribuídos são mais tolerantes a falhas. Assim, uma máquina cair não derruba todo o sistema. Consequentemente, a disponibilidade aumenta significativamente. Quando utilizar programação distribuída? Em problemas massivamente paralelos. Por exemplo, processamento de dados, machine learning ou busca em larga escala. Da mesma forma, qualquer tarefa que não caiba em uma máquina. Python oferece ferramentas como Celery, Dask, Ray e PySpark. Então, vamos explorar conceitos, padrões e exemplos práticos. Três subtítulos guiarão você pelo mundo da computação distribuída. Portanto, ao final, você saberá quando e como distribuir seu código.

Conceitos fundamentais: nós, comunicação e coordenação

Um nó é uma máquina individual no sistema distribuído. Nós se comunicam via rede usando mensagens ou chamadas RPC. A coordenação garante que todos trabalhem em direção ao mesmo objetivo. Quando usar arquitetura distribuída? Quando uma máquina é insuficiente. Também quando você precisa de alta disponibilidade. Por exemplo, serviços como Google ou Netflix usam milhares de nós. Exemplo básico de comunicação entre processos via socket:

# servidor_distribuido.py
import socket
import json
import threading
import time

def processar_tarefa(tarefa):
    """Processa uma tarefa recebida."""
    tipo = tarefa.get('tipo')
    dados = tarefa.get('dados', {})
    
    if tipo == 'soma':
        return sum(dados.get('numeros', []))
    elif tipo == 'multiplicacao':
        resultado = 1
        for n in dados.get('numeros', []):
            resultado *= n
        return resultado
    elif tipo == 'echo':
        return dados.get('mensagem', '')
    else:
        return f"Tipo desconhecido: {tipo}"

def tratar_cliente(conn, addr):
    """Trata um cliente conectado."""
    print(f"Cliente conectado: {addr}")
    try:
        while True:
            dados = conn.recv(4096)
            if not dados:
                break
            
            tarefa = json.loads(dados.decode())
            print(f"Recebida tarefa: {tarefa['tipo']} de {addr}")
            
            resultado = processar_tarefa(tarefa)
            resposta = json.dumps({'status': 'ok', 'resultado': resultado})
            conn.send(resposta.encode())
    except Exception as e:
        print(f"Erro com {addr}: {e}")
    finally:
        conn.close()
        print(f"Cliente {addr} desconectado")

def iniciar_servidor(host='localhost', porta=8888):
    """Inicia o servidor distribuído."""
    servidor = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    servidor.bind((host, porta))
    servidor.listen(5)
    print(f"Servidor distribuído rodando em {host}:{porta}")
    
    try:
        while True:
            conn, addr = servidor.accept()
            thread = threading.Thread(target=tratar_cliente, args=(conn, addr))
            thread.start()
    except KeyboardInterrupt:
        print("Servidor encerrado")
    finally:
        servidor.close()

if __name__ == "__main__":
    iniciar_servidor()

# cliente_distribuido.py
import socket
import json

def enviar_tarefa(host, porta, tarefa):
    """Envia uma tarefa para o servidor."""
    cliente = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        cliente.connect((host, porta))
        cliente.send(json.dumps(tarefa).encode())
        
        resposta = cliente.recv(4096)
        return json.loads(resposta.decode())
    finally:
        cliente.close()

if __name__ == "__main__":
    # Testando o servidor
    tarefas = [
        {'tipo': 'soma', 'dados': {'numeros': [1, 2, 3, 4, 5]}},
        {'tipo': 'multiplicacao', 'dados': {'numeros': [2, 3, 4]}},
        {'tipo': 'echo', 'dados': {'mensagem': 'Olá servidor!'}},
    ]
    
    for tarefa in tarefas:
        resultado = enviar_tarefa('localhost', 8888, tarefa)
        print(f"Tarefa {tarefa['tipo']}: {resultado}")

# servidor_distribuido.py

import socket

import json

import threading

import time

def processar_tarefa(tarefa):

"""Processa uma tarefa recebida."""

tipo = tarefa.get('tipo')

dados = tarefa.get('dados', {})

if tipo == 'soma':

return sum(dados.get('numeros', []))

elif tipo == 'multiplicacao':

resultado = 1

for n in dados.get('numeros', []):

resultado *= n

return resultado

elif tipo == 'echo':

return dados.get('mensagem', '')

else:

return f"Tipo desconhecido: {tipo}"

def tratar_cliente(conn, addr):

"""Trata um cliente conectado."""

print(f"Cliente conectado: {addr}")

try:

while True:

dados = conn.recv(4096)

if not dados:

break

tarefa = json.loads(dados.decode())

print(f"Recebida tarefa: {tarefa['tipo']} de {addr}")

resultado = processar_tarefa(tarefa)

resposta = json.dumps({'status': 'ok', 'resultado': resultado})

conn.send(resposta.encode())

except Exception as e:

print(f"Erro com {addr}: {e}")

finally:

conn.close()

print(f"Cliente {addr} desconectado")

def iniciar_servidor(host='localhost', porta=8888):

"""Inicia o servidor distribuído."""

servidor = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

servidor.bind((host, porta))

servidor.listen(5)

print(f"Servidor distribuído rodando em {host}:{porta}")

try:

while True:

conn, addr = servidor.accept()

thread = threading.Thread(target=tratar_cliente, args=(conn, addr))

thread.start()

except KeyboardInterrupt:

print("Servidor encerrado")

finally:

servidor.close()

if __name__ == "__main__":

iniciar_servidor()

# cliente_distribuido.py

import socket

import json

def enviar_tarefa(host, porta, tarefa):

"""Envia uma tarefa para o servidor."""

cliente = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

try:

cliente.connect((host, porta))

cliente.send(json.dumps(tarefa).encode())

resposta = cliente.recv(4096)

return json.loads(resposta.decode())

finally:

cliente.close()

if __name__ == "__main__":

# Testando o servidor

tarefas = [

{'tipo': 'soma', 'dados': {'numeros': [1, 2, 3, 4, 5]}},

{'tipo': 'multiplicacao', 'dados': {'numeros': [2, 3, 4]}},

{'tipo': 'echo', 'dados': {'mensagem': 'Olá servidor!'}},

]

for tarefa in tarefas:

resultado = enviar_tarefa('localhost', 8888, tarefa)

print(f"Tarefa {tarefa['tipo']}: {resultado}")

Esse exemplo mostra um nó servidor processando requisições. Vários clientes podem enviar tarefas simultaneamente. Na prática, usamos bibliotecas mais robustas que sockets puros.

Padrões de programação distribuída

Existem padrões comuns em sistemas distribuídos. O padrão mestre-escravo divide trabalho entre trabalhadores. O padrão pipeline processa dados em etapas sequenciais. O padrão publish-subscribe notifica múltiplos interessados. Quando usar cada padrão? Mestre-escravo para tarefas independentes. Pipeline para processamento de fluxo de dados. Pub-sub para eventos e notificações em tempo real. Além disso, cada padrão resolve problemas específicos. Exemplo simplificado do padrão mestre-escravo:

import multiprocessing
import time
import random

class Escravo:
    """Trabalhador que processa uma parte da tarefa."""
    
    def __init__(self, id_):
        self.id = id_
    
    def processar(self, dados):
        """Processa um lote de dados."""
        print(f"Escravo {self.id}: processando {len(dados)} itens")
        time.sleep(random.uniform(0.5, 1.5))  # Simula trabalho
        resultado = [x * 2 for x in dados]  # Exemplo: dobra os valores
        print(f"Escravo {self.id}: concluído")
        return resultado

class Mestre:
    """Coordenador que distribui trabalho entre escravos."""
    
    def __init__(self, num_escravos):
        self.escravos = [Escravo(i) for i in range(num_escravos)]
    
    def distribuir(self, dados, num_particoes=None):
        """Divide os dados e distribui para os escravos."""
        if num_particoes is None:
            num_particoes = len(self.escravos)
        
        # Divide os dados em partes
        tamanho_parte = len(dados) // num_particoes
        partes = []
        for i in range(num_particoes):
            inicio = i * tamanho_parte
            fim = inicio + tamanho_parte if i < num_particoes - 1 else len(dados)
            partes.append(dados[inicio:fim])
        
        # Distribui para os escravos (usando processos)
        with multiprocessing.Pool(processes=len(self.escravos)) as pool:
            resultados = pool.map(self._processar_escravo, 
                                 [(i, partes[i]) for i in range(len(partes))])
        
        # Combina os resultados
        resultado_final = []
        for r in resultados:
            resultado_final.extend(r)
        return resultado_final
    
    def _processar_escravo(self, args):
        idx, dados = args
        return self.escravos[idx].processar(dados)

# Exemplo de uso
if __name__ == "__main__":
    dados = list(range(100))  # 100 números
    print(f"Dados originais: {dados[:10]}...")
    
    mestre = Mestre(num_escravos=4)
    
    inicio = time.time()
    resultado = mestre.distribuir(dados)
    tempo = time.time() - inicio
    
    print(f"Resultado: {resultado[:10]}...")
    print(f"Tempo de processamento: {tempo:.2f}s")
    print(f"Verificação: {resultado == [x*2 for x in dados]}")

import multiprocessing

import time

import random

class Escravo:

"""Trabalhador que processa uma parte da tarefa."""

def __init__(self, id_):

self.id = id_

def processar(self, dados):

"""Processa um lote de dados."""

print(f"Escravo {self.id}: processando {len(dados)} itens")

time.sleep(random.uniform(0.5, 1.5)) # Simula trabalho

resultado = [x * 2 for x in dados] # Exemplo: dobra os valores

print(f"Escravo {self.id}: concluído")

return resultado

class Mestre:

"""Coordenador que distribui trabalho entre escravos."""

def __init__(self, num_escravos):

self.escravos = [Escravo(i) for i in range(num_escravos)]

def distribuir(self, dados, num_particoes=None):

"""Divide os dados e distribui para os escravos."""

if num_particoes is None:

num_particoes = len(self.escravos)

# Divide os dados em partes

tamanho_parte = len(dados) // num_particoes

partes = []

for i in range(num_particoes):

inicio = i * tamanho_parte

fim = inicio + tamanho_parte if i < num_particoes - 1 else len(dados)

partes.append(dados[inicio:fim])

# Distribui para os escravos (usando processos)

with multiprocessing.Pool(processes=len(self.escravos)) as pool:

resultados = pool.map(self._processar_escravo,

[(i, partes[i]) for i in range(len(partes))])

# Combina os resultados

resultado_final = []

for r in resultados:

resultado_final.extend(r)

return resultado_final

def _processar_escravo(self, args):

idx, dados = args

return self.escravos[idx].processar(dados)

# Exemplo de uso

if __name__ == "__main__":

dados = list(range(100)) # 100 números

print(f"Dados originais: {dados[:10]}...")

mestre = Mestre(num_escravos=4)

inicio = time.time()

resultado = mestre.distribuir(dados)

tempo = time.time() - inicio

print(f"Resultado: {resultado[:10]}...")

print(f"Tempo de processamento: {tempo:.2f}s")

print(f"Verificação: {resultado == [x*2 for x in dados]}")

O mestre divide o trabalho, e os escravos processam em paralelo. Esse padrão escala adicionando mais escravos (máquinas). Portanto, é um dos mais úteis na prática.

Ferramentas práticas: celery, dask e ray

Celery é o framework mais popular para filas de tarefas distribuídas. Ele usa um broker (Redis ou RabbitMQ) para comunicação. Dask é especializado em computação paralela para dados grandes. Ray é moderno e focado em machine learning distribuído. Quando usar cada ferramenta? Celery para tarefas assíncronas web. Dask para processamento de arrays e dataframes. Ray para treinamento de modelos e reinforcement learning. Além disso, cada uma tem sua comunidade e documentação. Exemplo conceitual com Celery (código parcial):

# Exemplo conceitual usando Celery (requer instalação)
# Arquivo: tasks.py
"""
from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task
def processar_dados(n):
    # Simula processamento pesado
    resultado = sum(i * i for i in range(n))
    return resultado

@app.task
def baixar_url(url):
    import requests
    resposta = requests.get(url)
    return resposta.status_code
"""

# Exemplo prático com Dask (mais fácil de testar)
import dask
from dask.distributed import Client
import time

def trabalho_pesado(x):
    """Função que será distribuída."""
    time.sleep(0.1)  # Simula trabalho
    return x * x

def demonstrar_dask():
    """Demonstra computação distribuída com Dask."""
    print("=== Dask - Computação Distribuída ===")
    
    # Inicia um cluster local (simula múltiplas máquinas)
    client = Client(n_workers=4, threads_per_worker=1)
    print(f"Dashboard disponível em: {client.dashboard_link}")
    
    # Cria tarefas atrasadas (lazy evaluation)
    tarefas = [dask.delayed(trabalho_pesado)(i) for i in range(20)]
    
    # Executa em paralelo
    inicio = time.time()
    resultados = dask.compute(*tarefas)
    tempo = time.time() - inicio
    
    print(f"Resultados: {resultados[:5]}...")
    print(f"Tempo com 4 workers: {tempo:.2f}s")
    
    # Comparação sequencial
    inicio_seq = time.time()
    resultados_seq = [trabalho_pesado(i) for i in range(20)]
    tempo_seq = time.time() - inicio_seq
    print(f"Tempo sequencial: {tempo_seq:.2f}s")
    print(f"Aceleração: {tempo_seq / tempo:.2f}x")
    
    client.close()

# Exemplo com Ray (moderno para ML)
def demonstrar_ray():
    """Demonstra Ray para computação distribuída."""
    try:
        import ray
        ray.init(ignore_reinit_error=True)
        
        @ray.remote
        def tarefa_ray(x):
            time.sleep(0.1)
            return x * x
        
        # Executa tarefas em paralelo
        inicio = time.time()
        futures = [tarefa_ray.remote(i) for i in range(20)]
        resultados = ray.get(futures)
        tempo = time.time() - inicio
        
        print(f"\n=== Ray - Computação Distribuída ===")
        print(f"Resultados: {resultados[:5]}...")
        print(f"Tempo: {tempo:.2f}s")
        
        ray.shutdown()
    except ImportError:
        print("\nRay não está instalado. Instale com: pip install ray")

if __name__ == "__main__":
    demonstrar_dask()
    demonstrar_ray()
    
    print("\n=== Dica: Para usar Celery ===")
    print("pip install celery redis")
    print("celery -A tasks worker --loglevel=info")

# Exemplo conceitual usando Celery (requer instalação)

# Arquivo: tasks.py

"""

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379')

@app.task

def processar_dados(n):

# Simula processamento pesado

resultado = sum(i * i for i in range(n))

return resultado

@app.task

def baixar_url(url):

import requests

resposta = requests.get(url)

return resposta.status_code

"""

# Exemplo prático com Dask (mais fácil de testar)

import dask

from dask.distributed import Client

import time

def trabalho_pesado(x):

"""Função que será distribuída."""

time.sleep(0.1) # Simula trabalho

return x * x

def demonstrar_dask():

"""Demonstra computação distribuída com Dask."""

print("=== Dask - Computação Distribuída ===")

# Inicia um cluster local (simula múltiplas máquinas)

client = Client(n_workers=4, threads_per_worker=1)

print(f"Dashboard disponível em: {client.dashboard_link}")

# Cria tarefas atrasadas (lazy evaluation)

tarefas = [dask.delayed(trabalho_pesado)(i) for i in range(20)]

# Executa em paralelo

inicio = time.time()

resultados = dask.compute(*tarefas)

tempo = time.time() - inicio

print(f"Resultados: {resultados[:5]}...")

print(f"Tempo com 4 workers: {tempo:.2f}s")

# Comparação sequencial

inicio_seq = time.time()

resultados_seq = [trabalho_pesado(i) for i in range(20)]

tempo_seq = time.time() - inicio_seq

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

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

client.close()

# Exemplo com Ray (moderno para ML)

def demonstrar_ray():

"""Demonstra Ray para computação distribuída."""

try:

import ray

ray.init(ignore_reinit_error=True)

@ray.remote

def tarefa_ray(x):

time.sleep(0.1)

return x * x

# Executa tarefas em paralelo

inicio = time.time()

futures = [tarefa_ray.remote(i) for i in range(20)]

resultados = ray.get(futures)

tempo = time.time() - inicio

print(f"\n=== Ray - Computação Distribuída ===")

print(f"Resultados: {resultados[:5]}...")

print(f"Tempo: {tempo:.2f}s")

ray.shutdown()

except ImportError:

print("\nRay não está instalado. Instale com: pip install ray")

if __name__ == "__main__":

demonstrar_dask()

demonstrar_ray()

print("\n=== Dica: Para usar Celery ===")

print("pip install celery redis")

print("celery -A tasks worker --loglevel=info")

Essas ferramentas abstraem a complexidade da rede. Você escreve código como se fosse local, mas ele roda distribuído. A fórmula da aceleração em sistemas distribuídos: \(S = \frac{T_{\text{seq}}}{T_{\text{dist}}} \approx N_{\text{máquinas}} \times E\) Onde E é a eficiência (geralmente 0.7 a 0.9). Portanto, programação distribuída é o próximo nível após multiprocessamento. Comece com Dask para dados e Ray para ML. Celery é ótimo para tarefas web assíncronas. Finalmente, distribua seu código e conquiste a escalabilidade horizontal.