无法pickle '_mysql_connector.MySQL'对象问题求助：多进程RS485数据处理

我在从RS485接收统计数据时，需要同时完成两个任务：将数据存入MySQL数据库，以及绘制统计图表。因为Python线程效率不够，改用多进程后出现了如下报错：

Traceback (most recent call last):
File "D:\GuZhangJianCe\shujukuchuanru2.py", line 296, in
p2.start()
File "D:\anaconda\envs\pytorch\lib\multiprocessing\process.py", line 121, in start
self._popen = self._Popen(self)
File "D:\anaconda\envs\pytorch\lib\multiprocessing\context.py", line 224, in _Popen
return _default_context.get_context().Process._Popen(process_obj)
File "D:\anaconda\envs\pytorch\lib\multiprocessing\context.py", line 327, in _Popen
return Popen(process_obj)
File "D:\anaconda\envs\pytorch\lib\multiprocessing\popen_spawn_win32.py", line 93, in init
reduction.dump(process_obj, to_child)
File "D:\anaconda\envs\pytorch\lib\multiprocessing\reduction.py", line 60, in dump
ForkingPickler(file, protocol).dump(obj)
TypeError: cannot pickle '_mysql_connector.MySQL' object

这个报错的核心原因是MySQL连接对象无法被pickle序列化，Windows下多进程启动时会序列化整个进程对象传递给子进程，如果你在主进程创建了数据库连接并传给子进程，就会触发这个错误。以下是两种可行的解决办法：

1. 子进程内独立创建数据库连接 #

不要在主进程初始化MySQL连接后传递给子进程，而是让每个需要操作数据库的子进程自行创建连接：

import multiprocessing
import mysql.connector

def save_to_db(data):
    # 子进程内单独初始化数据库连接
    conn = mysql.connector.connect(
        host='你的数据库地址',
        user='用户名',
        password='密码',
        database='目标数据库'
    )
    cursor = conn.cursor()
    # 执行数据插入操作
    insert_sql = "INSERT INTO 你的表名 (字段1, 字段2) VALUES (%s, %s)"
    cursor.execute(insert_sql, data)
    conn.commit()
    # 操作完成后关闭连接
    cursor.close()
    conn.close()

def plot_chart(data):
    # 这里写你的统计图表绘制逻辑
    pass

if __name__ == '__main__':
    # 模拟从RS485获取数据
    rs485_data = get_rs485_data()
    
    # 创建进程时只传递数据，不传递连接
    p1 = multiprocessing.Process(target=save_to_db, args=(rs485_data,))
    p2 = multiprocessing.Process(target=plot_chart, args=(rs485_data,))
    
    p1.start()
    p2.start()
    p1.join()
    p2.join()

2. 用进程队列传递数据，主进程/专用进程处理数据库操作 #

如果不想频繁创建销毁数据库连接，可以用multiprocessing.Queue传递数据，让单独的进程负责数据库写入，子进程只处理绘图：

import multiprocessing
import mysql.connector

def db_worker(data_queue):
    # 初始化一次数据库连接
    conn = mysql.connector.connect(
        host='你的数据库地址',
        user='用户名',
        password='密码',
        database='目标数据库'
    )
    cursor = conn.cursor()
    while True:
        data = data_queue.get()
        if data is None:  # 收到终止信号时退出循环
            break
        insert_sql = "INSERT INTO 你的表名 (字段1, 字段2) VALUES (%s, %s)"
        cursor.execute(insert_sql, data)
        conn.commit()
    # 结束时关闭连接
    cursor.close()
    conn.close()

def plot_chart(data):
    # 这里写你的统计图表绘制逻辑
    pass

if __name__ == '__main__':
    data_queue = multiprocessing.Queue()
    # 启动数据库专用处理进程
    db_process = multiprocessing.Process(target=db_worker, args=(data_queue,))
    db_process.start()
    
    # 循环从RS485获取数据
    while True:
        rs485_data = get_rs485_data()
        # 将数据放入队列，交给数据库进程处理
        data_queue.put(rs485_data)
        # 启动绘图进程处理当前数据
        plot_process = multiprocessing.Process(target=plot_chart, args=(rs485_data,))
        plot_process.start()
        plot_process.join()
    
    # 程序结束时发送终止信号，关闭数据库进程
    data_queue.put(None)
    db_process.join()

关键注意事项 #

• Windows下多进程代码必须放在if __name__ == '__main__':代码块内，避免重复初始化导致的异常
• 数据库连接本身是进程不安全的，绝对不能跨进程共享连接对象
• 数据量较大时，优先选择队列+专用进程的方案，减少连接创建销毁的性能损耗

内容的提问来源于stack exchange，提问作者不在藏身于荒野