如何用多线程独立处理SQLAlchemy+OracleDB的fetchmany数据块？

原代码核心问题 #

• 每次循环重复创建ThreadPoolExecutor，线程创建/销毁的开销抵消了并发收益，甚至拖慢速度
• 多线程共用同一个数据库cursor，oracledb的cursor并非线程安全，会导致数据异常或报错
• 任务粒度错误：把单一行数据作为并发任务，线程调度开销远大于处理数据的开销，完全发挥不出并发优势
• 分块处理逻辑不匹配：fetchmany返回的是多行数据的列表，却用map把每行单独传给处理方法，导致方法逻辑和输入不兼容

优化后的实现方案 #

核心思路 #

1. 线程池全局初始化，复用线程避免重复开销
2. 每个线程使用独立的数据库连接和cursor，彻底解决线程安全问题
3. 采用分片查询的方式，让每个线程负责查询并处理一部分数据，实现读取+处理的并行
4. 以整个数据分片为任务粒度，减少线程调度开销

代码实现 #

from concurrent.futures import ThreadPoolExecutor
from sqlalchemy import create_engine
from collections import defaultdict
import pandas as pd
import time

class ThreadedEngine:
    def __init__(self, db_url='oracle://MY_DB'):
        self.db_url = db_url
        # 初始化线程池，指定并发数（至少4个）
        self.executor = ThreadPoolExecutor(max_workers=4)

    def _process_chunk(self, chunk, columns):
        """将查询得到的行数据块转换为DataFrame所需的字典格式"""
        dict_object = defaultdict(list)
        for row in chunk:
            for col_idx, value in enumerate(row):
                dict_object[columns[col_idx]].append(value)
        return dict_object

    def _get_db_cursor(self):
        """每个线程独立获取数据库连接和cursor，避免线程安全问题"""
        engine = create_engine(self.db_url, pool_pre_ping=True)
        conn = engine.raw_connection()
        cursor = conn.cursor()
        cursor.arraysize = 10000
        cursor.prefetchrows = 1000000
        return conn, cursor

    def fetch_table(self, table_name, chunksize=10000):
        start = time.time()

        # 预获取表列名，避免每个线程重复查询
        conn, cursor = self._get_db_cursor()
        cursor.execute(f"SELECT * FROM {table_name} WHERE ROWNUM <= 1")
        columns = [desc[0] for desc in cursor.description]
        conn.close()

        # 获取表总条数，计算分片数量
        conn, cursor = self._get_db_cursor()
        cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
        total_rows = cursor.fetchone()[0]
        conn.close()
        num_chunks = (total_rows + chunksize - 1) // chunksize

        # 提交所有分片查询任务
        futures = []
        for i in range(num_chunks):
            offset = i * chunksize
            future = self.executor.submit(
                self._fetch_process_single_chunk,
                table_name, offset, chunksize, columns
            )
            futures.append(future)

        # 合并所有分片处理结果
        combined_dict = defaultdict(list)
        for future in futures:
            chunk_dict = future.result()
            for col, values in chunk_dict.items():
                combined_dict[col].extend(values)
        result_df = pd.DataFrame(combined_dict)

        end = time.time()
        print(f"总耗时: {end - start:.2f}秒")
        self.executor.shutdown()
        return result_df

    def _fetch_process_single_chunk(self, table_name, offset, chunksize, columns):
        """单个线程完成分片查询+数据处理的完整流程"""
        conn, cursor = self._get_db_cursor()
        try:
            # Oracle分页查询语法，避免全表扫描后分页
            query = """
                SELECT * FROM (
                    SELECT t.*, ROWNUM rn FROM {table} t
                ) WHERE rn > :offset AND rn <= :offset + :chunksize
            """.format(table=table_name)
            cursor.execute(query, {"offset": offset, "chunksize": chunksize})
            chunk_data = cursor.fetchall()
            return self._process_chunk(chunk_data, columns)
        finally:
            # 确保数据库连接关闭，避免资源泄漏
            conn.close()

使用示例 #

# 初始化引擎
engine = ThreadedEngine(db_url='oracle://USER:PASS@HOST:PORT/SERVICE')
# 分块并发读取表数据
df = engine.fetch_table("YOUR_TABLE_NAME", chunksize=10000)

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