多线程下pymssql.fetchall()性能骤降问题求助

• 线程限流逻辑完全失效：你的主线程仅在启动初期检查一次workers_online，随后直接批量启动所有查询线程，实际并发数等于main_list中的query数量，完全突破了max_workers=1的限制。大量并发线程同时拉取大结果集，直接导致本地网络带宽、内存资源被占满，每个线程的接收速度骤降。
• workers_online变量线程不安全：主线程对workers_online的递增操作未加锁，而子线程的递减操作仅用了锁，多线程环境下会出现竞态条件，该变量的数值完全不可信，根本无法起到限流作用。
• 大结果集一次性拉取的开销：cursor.fetchall()会一次性将所有查询结果加载到本地内存，10万行的结果集本身就需要占用不少内存，多线程同时拉取时，内存占用飙升，进一步拖慢数据接收速度。服务器端显示ASYNC_NETWORK_IO状态，本质是服务器已经处理完查询，正在等待客户端接收数据，而客户端因资源不足无法及时接收。
• Python GIL的间接影响：虽然IO操作会释放GIL，但多线程同时进行网络IO时，频繁的线程切换会产生额外开销，在大结果集场景下，这种开销会被放大，导致整体效率下降。

1. 使用线程池替代手动线程管理 #

手动管理线程容易出错，用concurrent.futures.ThreadPoolExecutor可以自动控制并发数，避免资源竞争：

from concurrent.futures import ThreadPoolExecutor
import threading
import pymssql

global_list = []
lock = threading.Lock()
max_workers = 4  # 根据本地资源调整，建议2-4

def get_query(my_query, var1, list1):
    connection = pymssql.connect(params)
    cursor = connection.cursor()
    cursor.execute(my_query)
    # 分批拉取结果，避免一次性加载大量数据
    batch_size = 1000
    records = []
    while True:
        batch = cursor.fetchmany(batch_size)
        if not batch:
            break
        records.extend(batch)
    cursor.close()
    connection.close()
    # 处理records逻辑
    with lock:
        global_list.append('some data')

def main():
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有查询任务
        futures = [executor.submit(get_query, query, var1, list1) for query in main_list]
        # 等待所有任务完成（可选，按需启用）
        for future in futures:
            try:
                future.result()
            except Exception as e:
                # 捕获任务异常
                print(f"任务执行出错: {e}")

if __name__ == "__main__":
    main()

2. 分批拉取结果集 #

将fetchall()替换为fetchmany()循环拉取，减少单次内存占用，缓解多线程下的内存竞争：

# 替换原fetchall()代码块
batch_size = 1000  # 可根据内存情况调整
records = []
while True:
    batch = cursor.fetchmany(batch_size)
    if not batch:
        break
    records.extend(batch)

3. 修复手动线程管理的线程安全问题（若坚持手动管理） #

如果不想用线程池，必须对workers_online的所有操作加锁，同时在每次启动线程前检查并发数：

def main():
    global workers_online
    for query in main_list:
        # 每次启动线程前检查并发数，确保不超过上限
        while True:
            with lock:
                if workers_online < max_workers:
                    workers_online += 1
                    break
        th = Thread(target=get_query, args=(query, var1, list1), daemon=False)
        th.start()

注意：子线程中的workers_online -=1已有锁，主线程的递增操作必须同样在锁内执行，避免竞态条件。

4. 使用数据库连接池 #

频繁创建销毁数据库连接会带来额外开销，使用连接池可以复用连接，提升效率：

# 示例：使用pymssql内置连接池
from pymssql import _mssql

# 初始化连接池，最大连接数与并发数匹配
pool = _mssql.ConnectionPool(server='your_server', user='username', password='pwd', database='your_db', maxconnections=max_workers)

def get_query(my_query, var1, list1):
    conn = pool.get_connection()
    cursor = conn.cursor()
    cursor.execute(my_query)
    # 分批拉取结果...
    cursor.close()
    pool.put_connection(conn)
    # 后续处理逻辑...

5. 合理调整并发数 #

不要盲目提高并发数，根据你的硬件配置（16GB内存、i5-11300H），建议将max_workers设置为2-4，避免资源过载导致的性能下降。

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