Geração de Dados Aleatórios e Inserção em Banco com Multithread e Multiprocessamento

Módulos Essenciais

string.ascii_letters: Constante do módulo string contendo letras ASCII maiúsculas e minúsculas. Requer importação: import string

string.digits: Constante com dígitos de 0 a 9. Requer import string

random.choice: Seleciona elemetno aleatório de sequências. Exemplos:

# Seleção de caracteres
import random
import string

caracteres = string.ascii_letters + string.digits
selecao = random.choice(caracteres)

# Seleção em lista
itens = ["maçã", "banana", "laranja"]
fruta_aleatoria = random.choice(itens)

Outras funções úteis:

  • random.randint(a, b): Inteiro aleatório em [a, b]
  • random.uniform(a, b): Float aleatório em [a, b]
  • random.sample(seq, k): Amostra sem repetição

Gerador de Dados

class GeradorDados:
    def __init__(self):
        self.nome = None
        self.contador = 0
    
    def gerar_nome(self):
        caracteres = string.ascii_letters + string.digits
        tamanho = random.randint(6, 12)
        self.nome = ''.join(random.choice(caracteres) for _ in range(tamanho))
        return self.nome
    
    def gerar_sexo(self):
        return random.choice(['Masculino', 'Feminino'])
    
    def gerar_email(self):
        return f"{self.nome}{random.choice(['@dom1.com', '@dom2.com'])}"
    
    def gerar_id(self, prefixo="ID"):
        self.contador += 1
        return f"{prefixo}{self.contador:05d}"
    
    def dados_completos(self):
        return [
            self.gerar_nome(),
            random.randint(18, 100),
            self.gerar_email(),
            self.gerar_id(),
            "Endereço Padrão",
            "Brasil",
            "Premium",
            "2024-01-01"
        ]

Inserção em Banco de Dados

class InseridorBanco:
    def __init__(self, conexao):
        self.conexao = conexao
        self.cursor = conexao.cursor()
        self.gerador = GeradorDados()
    
    def criar_tabela(self):
        schema = """
        CREATE TABLE IF NOT EXISTS usuarios (
            id INT AUTO_INCREMENT PRIMARY KEY,
            nome VARCHAR(50),
            idade INT,
            email VARCHAR(100),
            id_custom VARCHAR(20),
            endereco VARCHAR(100),
            pais VARCHAR(20),
            categoria VARCHAR(20),
            registro DATE
        )
        """
        self.cursor.execute(schema)
    
    def inserir_lote(self, dados):
        query = """
        INSERT INTO usuarios 
        (nome, idade, email, id_custom, endereco, pais, categoria, registro)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        """
        self.cursor.executemany(query, dados)
        self.conexao.commit()

Processamento Paralelo

Multithread

from concurrent.futures import ThreadPoolExecutor

def executar_threads(total_registros, tamanho_lote, num_threads):
    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        for _ in range(0, total_registros, tamanho_lote):
            lote = [gerador.dados_completos() for _ in range(tamanho_lote)]
            executor.submit(inseridor.inserir_lote, lote)

Multiprocessamento

from multiprocessing import Pool

def tarefa_processo(config, lote):
    inseridor = InseridorBanco(config)
    inseridor.inserir_lote(lote)

def executar_processos(total_registros, tamanho_lote, num_processos):
    lotes = [[gerador.dados_completos() for _ in range(tamanho_lote)] 
             for _ in range(0, total_registros, tamanho_lote)]
    
    with Pool(num_processos) as pool:
        pool.starmap(tarefa_processo, [(config, lote) for lote in lotes])

Medição de Desempenho

import time

def medir_desempenho(total, estrategia):
    inicio = time.time()
    estrategia()
    duracao = time.time() - inicio
    print(f"Tempo: {duracao:.2f}s | Velocidade: {total/duracao:.2f} registros/segundo")

Tags: Python MySQL Multithreading Multiprocessing geração-de-dados

Publicado em 7-17 13:52