多线程读取MySQL大表时出现连接丢失错误的解决求助

多线程读取MySQL中近7万行数据时，触发pymysql.OperationalError(2013) 'Lost connection to MySQL server during query'错误，同时伴随无法回滚的异常。但将LIMIT设为10000单线程一次性读取时，能正常获取全部数据。

代码实现 #

class Loader:
    """Load a table with multithreading"""
    def __init__(self, table: str, connector: Connector, fields: str=None, chunk_size: int=1000, query: str=None):
        # Config the table information
        self.fields = ['all' if fields is None else fields] # Query all the fields if user do NOT indicate.
        self.table = table
        self.conn = connector.conn
        self.cursor = connector.cursor

        self.cursor.execute(f"""SELECT COUNT(*) FROM {table}""")
        self.table_size = self.cursor.fetchone()[0]

        self.chunk_size = chunk_size
        self.num_chunks = self.table_size // self.chunk_size + (1 if self.table_size % self.chunk_size > 0 else 0)
        self.query = query # user's query

    def load_large_data(self):
        all_data = pd.DataFrame()  # Store all the data from SQL

        with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
            futures = [] # tasks queue
            # Commit multithreading tasks
            for i in range(self.num_chunks):
                offset = i * self.chunk_size
                futures.append(executor.submit(self.read_chunked_data, offset))

            # Collect the result after all the tasks are completed.
            for future in concurrent.futures.as_completed(futures):
                data_chunk = future.result()
                all_data = pd.concat([all_data, data_chunk], ignore_index=True)  # Merge tasks' data.

        return all_data

    def read_chunked_data(self, offset):
        # Execute embedded query either user's modified query.
        if self.query is None:
            if self.fields == 'all':
                query = f"""SELECT * FROM {self.table} LIMIT {self.chunk_size} OFFSET {offset}"""
            else:
                fields_string = ','.join(self.fields)
                query = f"""SELECT {fields_string} FROM {self.table} LIMIT {self.chunk_size} OFFSET {offset}"""
        else: # In case of 'WHERE' etc. is added by user.
            query = f"""{self.query} LIMIT {self.chunk_size} OFFSET {offset}"""

        return pd.read_sql(query, self.conn)

错误栈 #

Traceback (most recent call last):
  File "D:\Applications\Anaconda\Lib\site-packages\pandas\io\sql.py", line 2674, in execute
    cur.execute(sql, *args)
  File "D:\Applications\Anaconda\Lib\site-packages\pymysql\cursors.py", line 153, in execute
    result = self._query(query)
             ^^^^^^^^^^^^^^^^^^
  File "D:\Applications\Anaconda\Lib\site-packages\pymysql\cursors.py", line 322, in _query
    conn.query(q)
  File "D:\Applications\Anaconda\Lib\site-packages\pymysql\connections.py", line 563, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Applications\Anaconda\Lib\site-packages\pymysql\connections.py", line 825, in _read_query_result
    result.read()
  File "D:\Applications\Anaconda\Lib\site-packages\pymysql\connections.py", line 1199, in read
    first_packet = self.connection._read_packet()
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Applications\Anaconda\Lib\site-packages\pymysql\connections.py", line 753, in _read_packet
    raise err.OperationalError(
pymysql.err.OperationalError: (2013, 'Lost connection to MySQL server during query')

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "D:\Applications\Anaconda\Lib\site-packages\pandas\io\sql.py", line 2678, in execute
    self.con.rollback()
  File "D:\Applications\Anaconda\Lib\site-packages\pymysql\connections.py", line 492, in rollback
    self._execute_command(COMMAND.COM_QUERY, "ROLLBACK")
  File "D:\Applications\Anaconda\Lib\site-packages\pymysql\connections.py", line 843, in _execute_command
    raise err.InterfaceError(0, "")
pymysql.err.InterfaceError: (0, '')

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Users\whale\PycharmProjects\Risk Control\riskcontrol\test.py", line 33, in <module>
    result = loader.load_large_data()
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\whale\PycharmProjects\Risk Control\riskcontrol\database.py", line 96, in load_large_data
    data_chunk = future.result()
                 ^^^^^^^^^^^^^^^
  File "D:\Applications\Anaconda\Lib\concurrent\futures\_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "D:\Applications\Anaconda\Lib\concurrent\futures\_base.py", line 401, in __get_result
    raise self._exception
  File "D:\Applications\Anaconda\Lib\concurrent\futures\thread.py", line 58, in run
    result = self.fn(*args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\whale\PycharmProjects\Risk Control\riskcontrol\database.py", line 112, in read_chunked_data
    return pd.read_sql(query, self.conn)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Applications\Anaconda\Lib\site-packages\pandas\io\sql.py", line 706, in read_sql
    return pandas_sql.read_query(
           ^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Applications\Anaconda\Lib\site-packages\pandas\io\sql.py", line 2738, in read_query
    cursor = self.execute(sql, params)
             ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Applications\Anaconda\Lib\site-packages\pandas\io\sql.py", line 2683, in execute
    raise ex from inner_exc
pandas.errors.DatabaseError: Execution failed on sql: SELECT CONTENT FROM tb_customer_info LIMIT 1000 OFFSET 2000
(2013, 'Lost connection to MySQL server during query')
unable to rollback

核心原因 #

pymysql的Connection和Cursor对象并非线程安全，多个线程共享同一个连接时，会出现SQL请求交错、连接状态混乱的情况，导致MySQL服务器主动断开连接，触发2013错误；同时因为连接已中断，后续的回滚操作自然无法执行。

具体修复方案 #

1. 为每个线程创建独立数据库连接 #

不要在Loader类中共享同一个已建立的连接，改为让每个线程在执行查询时创建自己的连接实例，用完即关闭。修改后的代码如下：

class Loader:
    """Load a table with multithreading"""
    def __init__(self, table: str, connector_cls, db_config: dict, fields: str=None, chunk_size: int=1000, query: str=None):
        # 传入连接类和数据库配置，而非已建立的连接
        self.fields = ['all' if fields is None else fields]
        self.table = table
        self.connector_cls = connector_cls
        self.db_config = db_config

        # 临时连接获取表大小，用完关闭
        temp_conn = self.connector_cls(**self.db_config).conn
        temp_cursor = temp_conn.cursor()
        temp_cursor.execute(f"""SELECT COUNT(*) FROM {table}""")
        self.table_size = temp_cursor.fetchone()[0]
        temp_cursor.close()
        temp_conn.close()

        self.chunk_size = chunk_size
        self.num_chunks = self.table_size // self.chunk_size + (1 if self.table_size % self.chunk_size > 0 else 0)
        self.query = query

    def load_large_data(self):
        all_data = pd.DataFrame()

        with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
            futures = []
            for i in range(self.num_chunks):
                offset = i * self.chunk_size
                futures.append(executor.submit(self.read_chunked_data, offset))

            for future in concurrent.futures.as_completed(futures):
                data_chunk = future.result()
                all_data = pd.concat([all_data, data_chunk], ignore_index=True)

        return all_data

    def read_chunked_data(self, offset):
        # 每个线程创建独立连接
        connector = self.connector_cls(**self.db_config)
        conn = connector.conn

        if self.query is None:
            if self.fields == 'all':
                query = f"""SELECT * FROM {self.table} LIMIT {self.chunk_size} OFFSET {offset}"""
            else:
                fields_string = ','.join(self.fields)
                query = f"""SELECT {fields_string} FROM {self.table} LIMIT {self.chunk_size} OFFSET {offset}"""
        else:
            query = f"""{self.query} LIMIT {self.chunk_size} OFFSET {offset}"""

        data = pd.read_sql(query, conn)
        # 查询完成后关闭连接
        conn.close()
        return data

调用示例：

db_config = {
    'host': '你的数据库地址',
    'user': '用户名',
    'password': '密码',
    'database': '数据库名'
}
loader = Loader('tb_customer_info', Connector, db_config, fields='CONTENT', chunk_size=1000)

2. 优化LIMIT OFFSET性能（可选） #

LIMIT OFFSET在偏移量较大时，MySQL需要扫描大量无关数据，效率低下。可以改用基于主键的范围查询，避免OFFSET的性能损耗：

# 假设表有自增主键id，先获取主键范围再分块查询
def read_chunked_data(self, start_id, end_id):
    connector = self.connector_cls(**self.db_config)
    conn = connector.conn
    query = f"""SELECT CONTENT FROM {self.table} WHERE id >= {start_id} AND id < {end_id}"""
    data = pd.read_sql(query, conn)
    conn.close()
    return data

3. 调整MySQL连接超时参数（辅助） #

如果服务器端连接超时设置过短，可调整MySQL配置：

• 修改my.cnf（或my.ini）中的wait_timeout和interactive_timeout，设置为更大的值（例如300秒）
• 建立连接时，通过connect方法传入connect_timeout参数，延长连接超时时间

验证 #

修改后运行多线程读取，每个线程使用独立连接，不会出现连接共享导致的冲突，即可解决2013错误和回滚失败问题。

内容的提问来源于stack exchange，提问作者Truman