Python多线程爬虫写入MySQL速度特别慢如何优化

我遇到过多线程爬虫写MySQL慢的问题，核心瓶颈通常是数据库连接和事务处理。给你一个优化方案：

import threading
import queue
import pymysql
from pymysql import cursors
from concurrent.futures import ThreadPoolExecutor
import time

class MySQLBatchWriter:
    def __init__(self, host, user, password, database, batch_size=1000, max_workers=5):
        self.batch_size = batch_size
        self.data_queue = queue.Queue(maxsize=5000)
        self.lock = threading.Lock()
        self.connection_pool = []
        
        # 创建连接池
        for _ in range(max_workers):
            conn = pymysql.connect(
                host=host,
                user=user,
                password=password,
                database=database,
                charset='utf8mb4',
                cursorclass=cursors.DictCursor,
                autocommit=False  # 关闭自动提交
            )
            self.connection_pool.append(conn)
        
        # 启动写入线程
        self.writer_thread = threading.Thread(target=self._batch_writer, daemon=True)
        self.writer_thread.start()
    
    def add_data(self, data_dict):
        """生产者线程调用此方法添加数据"""
        self.data_queue.put(data_dict)
    
    def _batch_writer(self):
        """专门的写入线程，批量提交"""
        batch = []
        conn_index = 0
        
        while True:
            try:
                # 从队列获取数据
                data = self.data_queue.get(timeout=1)
                batch.append(data)
                
                # 达到批量大小或队列为空时写入
                if len(batch) >= self.batch_size or self.data_queue.empty():
                    if batch:
                        self._insert_batch(batch, conn_index)
                        batch = []
                        conn_index = (conn_index + 1) % len(self.connection_pool)
                        
            except queue.Empty:
                if batch:
                    self._insert_batch(batch, conn_index)
                    batch = []
                time.sleep(0.1)
    
    def _insert_batch(self, batch, conn_index):
        """批量插入数据"""
        conn = self.connection_pool[conn_index]
        try:
            with conn.cursor() as cursor:
                # 构建批量插入SQL
                placeholders = ', '.join(['%s'] * len(batch[0]))
                columns = ', '.join(batch[0].keys())
                sql = f"INSERT INTO your_table ({columns}) VALUES ({placeholders})"
                
                # 准备数据
                values = [tuple(item.values()) for item in batch]
                
                # 执行批量插入
                cursor.executemany(sql, values)
                conn.commit()
                
                print(f"批量插入 {len(batch)} 条数据成功")
                
        except Exception as e:
            conn.rollback()
            print(f"批量插入失败: {e}")
            # 这里可以添加重试或错误处理逻辑
    
    def close(self):
        """关闭所有连接"""
        for conn in self.connection_pool:
            conn.close()

# 使用示例
def crawler_worker(writer, page_range):
    """爬虫工作线程"""
    for page in page_range:
        # 模拟爬取数据
        data = {
            'title': f'page_{page}',
            'url': f'http://example.com/page/{page}',
            'content': '...'
        }
        writer.add_data(data)

# 主程序
if __name__ == '__main__':
    # 创建批量写入器
    writer = MySQLBatchWriter(
        host='localhost',
        user='root',
        password='password',
        database='test_db',
        batch_size=500,  # 每500条批量提交一次
        max_workers=3    # 3个数据库连接
    )
    
    # 创建爬虫线程池
    with ThreadPoolExecutor(max_workers=10) as executor:
        # 分配爬取任务
        page_ranges = [(i*100, (i+1)*100) for i in range(10)]
        futures = []
        
        for start, end in page_ranges:
            future = executor.submit(crawler_worker, writer, range(start, end))
            futures.append(future)
        
        # 等待所有爬虫任务完成
        for future in futures:
            future.result()
    
    # 等待队列中的数据全部写入
    time.sleep(2)
    writer.close()

关键优化点：

1. 批量写入：使用executemany()批量插入，减少网络往返
2. 连接池：复用数据库连接，避免频繁创建连接的开销
3. 生产者-消费者模式：爬虫线程只负责生产数据，专门的写入线程负责批量提交
4. 关闭自动提交：手动控制事务提交，批量操作完成后一次性提交
5. 队列缓冲：使用队列解耦爬取和写入，避免相互阻塞

这样改造后，写入速度能提升几十倍。记得根据实际情况调整batch_size和连接数。

请教一下，服务器中跑的服务很多，瞬息万变，可能一转眼系统就开始杀进程了，该如何把握时机把数据存入数据库呢？

对于中小量数据,直接用 pandas
pd.Dataframe([]) 暂存内存后直接 to_sql,大量数据 用 scrapy+mysql/mongodb