Python如何使用全部CPU核心而非线程实现多进程数据库操作

multiprocessing.dummy是对threading模块的封装，实现的是线程池，受Python GIL限制无法利用多核CPU做并行计算，要调用全部CPU核心直接替换为原生multiprocessing模块的进程池即可，两者API完全兼容，不需要改动池的调用逻辑。

注意你的现有代码存在几个会导致多进程运行异常、性能严重损耗的问题，需要同步修正：

• 数据库连接、SQLAlchemy引擎对象不能跨进程传递，必须在每个子进程的任务函数内部初始化，这点你现有代码的写法是对的，但要注意用完主动释放连接，避免连接泄漏。
• 现有逻辑无论传入什么参数，都会重复执行全表查询+全量写入，多进程跑只会重复拉取重复写入，完全浪费资源，必须把查询条件和传入的分片参数绑定，按维度拆分查询任务，才能真正发挥并行优势。
• echo=True会打印全量SQL执行日志，高并发场景下日志IO会占用大量CPU资源，非调试场景直接关闭。
• 进程数不要直接拉满cpu_count()，数据库IO场景下进程数设置为CPU核心数的1~2倍即可，进程过多会打满数据库连接、引发锁冲突，反而降低整体效率。
• to_sql写入时添加chunksize参数，避免大结果集一次性写入占用过多内存，同时设置index=False避免把pandas的默认索引写入表中产生脏数据。

修正后可直接运行的代码 #

import pandas as pd
from sqlalchemy import create_engine
# 替换dummy线程池为原生多进程池
import multiprocessing as mp
from multiprocessing import cpu_count

# 按查询维度拆分任务参数，示例为按订单日期分片
dates_to_extract = ['2024-01-01', '2024-01-02', '2024-02-01']
# 数据库连接串抽为常量，避免重复硬编码
DB_CONN = "mysql+pymysql://kieran:cccer@localhost/ran"

def run_query(query_date):
    source_conn = None
    dest_conn = None
    source_engine = None
    dest_engine = None
    try:
        # 每个子进程独立初始化引擎和连接
        source_engine = create_engine(DB_CONN, echo=False)
        dest_engine = create_engine(DB_CONN, echo=False)
        source_conn = source_engine.connect()
        dest_conn = dest_engine.connect()

        print(f"start processing date: {query_date}")
        # 按分片参数拼接查询条件，避免重复查全表
        sql = f"SELECT * FROM ran.order_data WHERE order_date = '{query_date}'"
        df = pd.read_sql(sql, source_conn)

        # 分批写入目标表
        df.to_sql(
            name='tbjd',
            con=dest_conn,
            if_exists='append',
            index=False,
            chunksize=1000
        )
        print(f"finish processing date: {query_date}")
    except Exception as e:
        print(f"process {query_date} failed: {str(e)}")
    finally:
        # 主动释放连接，避免连接泄漏
        if source_conn:
            source_conn.close()
        if dest_conn:
            dest_conn.close()
        if source_engine:
            source_engine.dispose()
        if dest_engine:
            dest_engine.dispose()

if __name__ == "__main__":
    # 进程数取CPU核心数2倍和8的较小值，可根据数据库最大连接数调整
    worker_num = min(cpu_count() * 2, 8)
    pool = mp.Pool(worker_num)
    pool.map(run_query, dates_to_extract)
    pool.close()
    pool.join()

额外优化建议 #

• 数据库查询属于IO密集型操作，多进程存在进程启动、数据序列化的额外开销，实际效率不一定比多线程高，建议同并发数下对比测试两种模式的耗时，选择性能更好的方案即可。
• 大数据量写入时可以暂时关闭目标表的二级索引、事务自动提交，写入完成后再重建索引，能大幅提升写入速度。
• 如果单表数据量超过千万级，建议在分片查询时再加一层主键范围分页，避免单次查询拉取过大结果集撑爆内存。

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