Escrita Concorrente Segura em Excel com Openpyxl e ThreadPoolExecutor
Entendendo o Conceito
O gerenciamento manual do ciclo de vida das threads (como criação, ativação e junção) pode se tornar complexo e propenso a erros à medida que o número de tarefas concorrentes cresce. A classe ThreadPoolExecutor, integrada ao módulo concurrent.futures, simplifica essa administração ao fornecer um pool de threads trabalhadoras pré-configuradas para executar tarefas de entrada e saída (I/O) de forma assíncrona. Essa abordagem otimiza o uso de recursos e simplifica a estrutura do código concorrente.
Contudo, o uso de um executor em pool não elimina a necessidade de controle de concorrência nos objetos de planilha compartilhados. Como o openpyxl continua sendo não seguro para threads (not thread-safe), as requisições enviadas ao pool de threads podem corromper as estruturas internas do arquivo Excel se acessarem o mesmo objeto simultaneamente. Por essa razão, a sincronização utilizando travas de exclusão mútua (locks) continua obrigatória para proteger o bloco crítico de gravação e gravação física do arquivo em disco.
Passo a Passo
O processo de implementação envolve configurar um pool de threads que delega tarefas para uma classe gerenciadora equipada com travas de segurança.
Primeiramente, certifique-se de instalar a biblioteca openpyxl:
pip install openpyxl
Em seguida, implemente o script contendo a classe protetora e o uso do gerenciador de contexto ThreadPoolExecutor:
import threading
from concurrent.futures import ThreadPoolExecutor
from openpyxl import Workbook
class ThreadSafeExcelWriter:
def __init__(self, filename):
self.filename = filename
self.wb = Workbook()
self.ws = self.wb.active
self.lock = threading.Lock()
def write_data(self, row, col, value):
with self.lock:
self.ws.cell(row=row, column=col, value=value)
def save_workbook(self):
with self.lock:
self.wb.save(self.filename)
def worker_task(writer, start_row, value):
for i in range(5):
writer.write_data(row=start_row + i, col=1, value=value)
writer = ThreadSafeExcelWriter("dados_concorrentes.xlsx")
with ThreadPoolExecutor(max_workers=4) as executor:
futures = [
executor.submit(worker_task, writer, index * 10, f"Valor-{index}")
for index in range(1, 6)
]
writer.save_workbook()
Explicando o código
A rotina de execução estruturada implementa conceitos específicos para o tratamento de chamadas paralelas:
ThreadPoolExecutor(max_workers=4): Define um pool limitado a um máximo de quatro threads operando em paralelo para processar as escritas concorrentes.with ThreadPoolExecutor(...) as executor: Gerenciador de contexto que assegura a interrupção adequada do pool e realiza o join automático das threads, bloqueando a execução posterior até que todas as tarefas agendadas tenham sido integralmente finalizadas.executor.submit(worker_task, ...): Envia a tarefa de forma não bloqueante para o pool de threads e retorna imediatamente uma referência Future correspondente àquela execução.threading.Lock(): Instancia uma trava de controle de concorrência que impede o acesso concorrente aos métodos mutáveis do openpyxl [12, 16].with self.lock:: Gerenciador que delimita a seção crítica do código, acionando a trava e garantindo sua posterior liberação automática, mesmo diante de exceções no decorrer do processamento de dados [13, 15].self.ws.cell(row=row, column=col, value=value): Grava as informações nas coordenadas de linha e coluna da planilha de maneira segura sob o lock ativo.
Observações
- A adoção do ThreadPoolExecutor reduz significativamente o código boilerplate em comparação ao controle manual de loops para criação e junção de threads do sistema operacional.
- Configurar o parâmetro max_workers com valores excessivamente altos para tarefas concorrentes de planilha gera forte contenção de trava (lock contention), anulando os benefícios de desempenho devido à constante espera pela liberação do recurso de gravação.
- O salvamento definitivo do arquivo deve ser chamado estritamente fora do escopo do bloco with do ThreadPoolExecutor, garantindo que o disco só seja gravado quando nenhuma thread do pool estiver ativa.
- Exceções geradas dentro de tarefas do pool de threads não são exibidas de imediato no console principal; elas ficam encapsuladas nos objetos Future e apenas se tornam visíveis ao chamar explicitamente o método
.result()de cada futuro. - Para gerenciar grandes cargas de dados, pode-se mapear listas de parâmetros diretamente para o pool por meio do método
executor.map, reduzindo o volume de chamadas individuais do método submit.
Referências
- Documentação de Concorrência e ThreadPoolExecutor no Python: https://docs.python.org/3/library/concurrent.futures.html
- Práticas Recomendadas do ThreadPoolExecutor no SuperFastPython: https://superfastpython.com/threadpoolexecutor-best-practices/
- Guia de Thread Safety e Sincronização da Real Python: https://realpython.com/python-thread-lock/
- Controle e Manipulação do openpyxl de Modo Multithread: https://stackoverflow.com/questions/11234203/openpyxl-thread-safe