Flask+SocketIO多进程报错：Can't pickle local object问题求助

你遇到的AttributeError: Can't pickle local object 'face_recognition_multi_core..one_to_one'错误，核心原因是Python的pickle模块没办法序列化嵌套在函数内部的局部函数。ProcessPoolExecutor启动子进程时，需要把任务函数和参数序列化后传递给子进程，但局部函数依赖父函数的上下文环境，pickle无法完整保存这些上下文信息，所以直接抛出了这个错误。

具体修复步骤 #

1. 将one_to_one函数移到全局作用域 #

把原来嵌套在face_recognition_multi_core里的one_to_one函数提取到模块顶层，让它成为全局函数——这样pickle就能正常序列化它了。

2. 显式传递依赖参数到子进程 #

多进程之间内存是完全隔离的，子进程没法直接访问主进程中的变量（比如FREQUENCY、TOLERANCE_THRES、faces_encodings），所以必须把这些变量作为参数直接传给one_to_one，不能让它依赖外部作用域的变量。

3. 修正SocketIO的跨进程使用 #

注意：子进程里不能直接调用主进程的socket_io.emit，因为SocketIO的连接绑定在主进程的事件循环中。推荐用「子进程写结果到队列，主进程开线程监听队列并发送消息」的方式解决，后面会给出具体代码。

修复后的核心代码 #

import time
import cv2
import dlib
import pickle
from datetime import datetime
from flask import Flask, request, render_template
from flask_socketio import SocketIO
from concurrent.futures import ProcessPoolExecutor, wait
from threading import Thread, Event
from multiprocessing import Queue

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
app.config['DEBUG'] = True
socket_io = SocketIO(app, async_mode=None, logger=True, engineio_logger=True)
thread = Thread()
thread_stop_event = Event()
# 全局队列用于子进程和主进程的结果通信
result_queue = Queue()

# 把one_to_one移到全局作用域，所有依赖参数都显式传入
def one_to_one(stream_link, count, frequency, tolerance_thres, faces_encodings):
    cap = cv2.VideoCapture(stream_link)
    if not cap.isOpened():
        print(f"Error opening video stream: {stream_link}")
        return
    try:
        # 子进程内部初始化检测器，避免跨进程序列化问题
        hogFaceDetector = dlib.get_frontal_face_detector()
        while cap.isOpened():
            count += 1
            ret, frame = cap.read()
            if not ret:
                break
            frame = resize_img(frame, 50)
            if count % frequency == 0:
                face_locations = hogFaceDetector(frame, 0)
                if len(face_locations) > 0:
                    sorted_detections_distances = calculate_face_distance(face_locations, frame, faces_encodings)
                    if sorted_detections_distances[0][1] < tolerance_thres:
                        now = datetime.now()
                        found_date = now.strftime('%m/%d/%Y')
                        result = {
                            "Location": stream_link,
                            "worker_name": sorted_detections_distances[0][0],
                            "time": found_date
                        }
                        # 多进程下不能直接修改全局csv_data，改用文件同步
                        try:
                            with open('multi_core.pickle', 'rb') as handle:
                                current_csv = pickle.load(handle)
                        except FileNotFoundError:
                            current_csv = []
                        current_csv.append(result)
                        with open('multi_core.pickle', 'wb') as handle:
                            pickle.dump(current_csv, handle, protocol=pickle.HIGHEST_PROTOCOL)
                        # 把结果放入队列，由主进程线程发送SocketIO消息
                        result_queue.put(result)
            if cv2.waitKey(25) & 0xFF == ord('q'):
                break
    finally:
        cap.release()
        cv2.destroyAllWindows()

# 后台线程：监听队列并发送SocketIO消息
def socketio_emitter():
    while not thread_stop_event.is_set():
        if not result_queue.empty():
            result = result_queue.get()
            socket_io.emit('my_response', {"data": dict(result)})
        time.sleep(0.1)

# 客户端连接时启动后台线程
@socket_io.on('connect')
def on_connect():
    global thread
    if not thread.is_alive():
        thread = Thread(target=socketio_emitter)
        thread.start()

@app.route('/face_reognition/multi_core', methods=["GET", "POST"])
def face_recognition_multi_core():
    try:
        start = time.time()
        if request.method == "POST":
            stream_links = request.form['stream_links']
            # 处理参数，设置默认值
            frequency = int(request.form.get('frequency', 10))
            tolerance_thres = float(request.form.get('tolerance_threshold', 0.45))
            
            # 提前加载人脸编码，避免每个子进程重复加载
            with open('faces_encodings.pickle', 'rb') as handle:
                faces_encodings = pickle.load(handle)
            
            # 初始化结果文件（如果不存在）
            try:
                with open('multi_core.pickle', 'rb'):
                    pass
            except FileNotFoundError:
                with open('multi_core.pickle', 'wb') as handle:
                    pickle.dump([], handle)
            
            # 启动多进程处理
            with ProcessPoolExecutor() as executor:
                futures = []
                count = 0
                for stream_link in stream_links:
                    futures.append(executor.submit(
                        one_to_one, 
                        stream_link, 
                        count, 
                        frequency, 
                        tolerance_thres, 
                        faces_encodings
                    ))
                # 可选：等待所有任务完成，根据业务需求决定是否需要
                wait(futures)
            
            end = time.time() - start
            print(f"总处理耗时: {end:.2f}秒")
            return render_template("index.html")
    except Exception as e:
        return str(e)

额外注意事项 #

• 多进程下全局变量csv_data无法共享，所以必须用文件、队列或multiprocessing.Manager来同步数据，上面的代码用了文件方式实现。
• 每个子进程会独立初始化dlib检测器，这是多进程内存隔离的正常现象，也可以避免跨进程序列化检测器的问题。

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