Gerenciamento de Tarefas Concorrentes com concurrent.futures em Python

O módulo concurrent.futures fornece uma API de alto nível para execução paralela de tarefas, simplificando a criação de threads e processos. Ele abstrai detalhes de threads e processos, permitindo focar na lógica do programa, ideal para I/O-bound e CPU-bound.

---

1. Conceitos Fundamentais

1.1 Executor

• Executor é a interface que gerencia threads ou processos.

• Dois tipos principais:

  • ThreadPoolExecutor → threads (I/O-bound)

  • ProcessPoolExecutor → processos (CPU-bound)

from concurrent.futures import ThreadPoolExecutor

def tarefa(x):
    return x**2

with ThreadPoolExecutor(max_workers=5) as executor:
    resultados = list(executor.map(tarefa, range(10)))
print(resultados)

• max_workers define o número máximo de threads/processos simultâneos.

---

1.2 Futures

• Um Future é um objeto que representa o resultado de uma execução que pode ainda não ter terminado.

from concurrent.futures import ThreadPoolExecutor

def tarefa(x):
    return x**2

executor = ThreadPoolExecutor(max_workers=3)
future = executor.submit(tarefa, 10)
print(future.result())  # bloqueia até o resultado estar pronto
executor.shutdown()

• submit() agenda a execução de uma função e retorna um Future.

• result() retorna o resultado da função, bloqueando até estar disponível.

---

2. Estratégias Avançadas de Gerenciamento

2.1 Map vs Submit

• executor.map(func, iterável) → executa todas as tarefas e retorna resultados na ordem do iterável.

• executor.submit(func, *args) → retorna Future individual para cada chamada, permitindo maior controle e flexibilidade.

from concurrent.futures import as_completed

executor = ThreadPoolExecutor(max_workers=5)
futures = [executor.submit(tarefa, i) for i in range(10)]

for future in as_completed(futures):
    print(future.result())  # resultados na ordem de conclusão

• as_completed permite processar resultados à medida que ficam prontos, útil para tarefas com duração variável.

---

2.2 Cancelamento de Tarefas

• Futures podem ser cancelados antes de iniciar a execução:

future = executor.submit(tarefa, 20)
if not future.done():
    future.cancel()

• Útil em pipelines dinâmicos ou timeout.

---

2.3 Timeout em Futures

try:
    resultado = future.result(timeout=2)
except TimeoutError:
    print("A tarefa excedeu o tempo limite")

• Permite gerenciar tarefas que podem travar ou demorar demais.

---

3. ThreadPoolExecutor vs ProcessPoolExecutor

3.1 ThreadPoolExecutor (I/O-bound)

• Ideal para requisições de rede, leitura/escrita de arquivos e scraping.

• Threads compartilham mesmo espaço de memória, fácil de comunicar dados.

• Limitado em tarefas CPU-bound devido ao GIL.

3.2 ProcessPoolExecutor (CPU-bound)

• Cada processo tem memória própria, ideal para cálculos pesados.

• Pode paralelizar funções de CPU intensivo sem GIL.

• Overhead maior ao criar processos, mas ganha performance em tarefas longas.

---

4. Técnicas Avançadas

4.1 Pipeline Produtor-Consumidor

import queue
from concurrent.futures import ThreadPoolExecutor

fila = queue.Queue()

def produtor():
    for i in range(10):
        fila.put(i)

def consumidor():
    while not fila.empty():
        item = fila.get()
        print(f"Consumiu {item}")

with ThreadPoolExecutor(max_workers=4) as executor:
    executor.submit(produtor)
    for _ in range(3):
        executor.submit(consumidor)

• Combina ThreadPoolExecutor com filas thread-safe para pipelines concorrentes.

4.2 Execução Condicional e Encadeada

def tarefa_condicional(x):
    if x % 2 == 0:
        return x**2
    return x

with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(tarefa_condicional, range(10)))

• Permite decisões por tarefa, mantendo concorrência eficiente.

---

4.3 Combinação de ProcessPool e ThreadPool

• Em pipelines híbridos:

  • ThreadPool para tarefas I/O-bound

  • ProcessPool para cálculos CPU-bound

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

def io_task(x):
    return x*2

def cpu_task(x):
    total = 0
    for i in range(1000000):
        total += i*x
    return total

with ThreadPoolExecutor() as t_executor, ProcessPoolExecutor() as p_executor:
    io_results = list(t_executor.map(io_task, range(5)))
    cpu_results = list(p_executor.map(cpu_task, io_results))

• Estratégia avançada para pipelines complexos e escaláveis.

---

5. Boas Práticas Profissionais

1. Prefira map para tarefas simples e submit/as_completed para maior controle.

2. Use timeout e cancelamento para evitar travamentos em tarefas longas.

3. Combine ThreadPoolExecutor para I/O e ProcessPoolExecutor para CPU-bound.

4. Evite compartilhar dados mutáveis entre processos; use queues ou managers.

5. Monitore Futures e status para rastrear execução e performance.

6. Limite max_workers de acordo com recursos disponíveis (CPU e memória).

---

6. Aplicações Profissionais

• Web scraping massivo com threads

• Chamadas concorrentes a APIs externas

• ETL e pipelines de dados paralelos

• Processamento de imagens e cálculos científicos

• Sistemas de monitoramento e coleta de dados

---

7. Conclusão

• concurrent.futures simplifica a execução paralela em Python.

• ThreadPoolExecutor é ideal para I/O, ProcessPoolExecutor para CPU-bound.

• Técnicas avançadas incluem as_completed, timeout, cancelamento, pipelines híbridos e combinação de threads/processos.

• Seguindo boas práticas, é possível criar aplicações escaláveis, rápidas e seguras para cenários profissionais complexos.