### 高效生成1亿条数据并批量插入MySQL的解决方案 #### 数据生成与插入策略 为了高效生成1亿条数据并插入到 MySQL 数据库中，可以采用多线程、批量插入以及事务管理等技术。以下是一个完整的实现方案[^1]。 #### 代码实现 ```python import pymysql import threading from queue import Queue # 数据库连接配置 host = "127.0.0.1" port = 3306 username = "root" password = "123" charset = "utf8mb4" db = "test" # 插入数据的函数 def insert_data(queue, connection): cursor = connection.cursor() while not queue.empty(): data_chunk = queue.get() sql = "INSERT INTO products (name, price) VALUES (%s, %s)" try: cursor.executemany(sql, data_chunk) connection.commit() except Exception as e: connection.rollback() print(f"Error occurred: {e}") finally: queue.task_done() # 主程序逻辑 def main(): # 创建数据库连接池 connections = [pymysql.connect( host=host, user=username, password=password, db=db, port=port, charset=charset ) for _ in range(5)] # 假设使用5个线程 # 创建任务队列 queue = Queue() # 生成1亿条数据并分批放入队列 total_records = 100_000_000 sql_batch = 10_000 # 每次commit的数据量 for i in range(0, total_records, sql_batch): data_chunk = [(f"Product_{j}", j * 0.1) for j in range(i, min(i + sql_batch, total_records))] queue.put(data_chunk) # 启动多线程插入 threads = [] for connection in connections: thread = threading.Thread(target=insert_data, args=(queue, connection)) thread.start() threads.append(thread) # 等待所有线程完成 for thread in threads: thread.join() # 关闭所有数据库连接 for connection in connections: connection.close() if __name__ == "__main__": main() ``` #### 实现细节说明 - **多线程**：通过创建多个线程并分配不同的数据库连接，实现并发插入以提高效率[^1]。 - **批量插入**：每次插入 `sql_batch` 条记录，减少与数据库交互的次数，从而提升性能[^1]。 - **任务队列**：使用 `Queue` 管理数据块，确保每个线程能够按需获取数据进行处理。 - **事务管理**：在插入失败时回滚事务，保证数据一致性。 #### 性能优化建议 - **调整线程数**：根据服务器硬件资源（如 CPU 核心数）合理设置线程数，避免因线程过多导致的上下文切换开销。 - **增大批量大小**：适当增加 `sql_batch` 的值，但需注意内存消耗和数据库的最大包大小限制[^1]。 - **索引管理**：在大规模数据插入前，临时禁用表上的索引，插入完成后重新启用，以减少插入过程中的索引维护开销[^2]。 #### 注意事项 - 确保数据库有足够的存储空间来容纳1亿条记录。 - 如果数据生成逻辑复杂，可以考虑将数据生成部分分离为独立模块，以便于调试和扩展[^2]。