Python多进程池批量POST结果：如何为每个Worker执行后处理？

核心思路 #

利用多进程池的进程初始化钩子和进程退出自动回调，给每个Worker进程绑定专属的结果缓存和退出后处理逻辑，确保每个Worker结束时自动提交剩余未批量的数据，无需全局队列或定时器。

具体实现步骤 #

1. 进程本地缓存初始化：通过multiprocessing.Pool的initializer参数，给每个Worker进程初始化一个进程专属的结果缓存（用multiprocessing.local()实现，避免进程间数据共享）。
2. 注册退出后处理函数：在初始化函数中，用atexit模块给当前Worker进程注册退出回调函数，进程结束时自动触发，提交缓存中剩余的结果。
3. 任务处理逻辑：在任务函数中，将结果加入进程本地缓存，攒够指定数量（比如3条）就立即批量提交并清空缓存。

代码示例 #

import multiprocessing
import atexit
import requests

# 进程本地存储，每个Worker进程有独立的实例
local_data = multiprocessing.local()

BATCH_SIZE = 3
POST_URL = "https://your-api-endpoint.com/submit"

def init_worker():
    # 初始化当前Worker的结果缓存
    local_data.gathered_data = []
    # 注册进程退出时的后处理函数
    atexit.register(post_remaining_data)

def post_remaining_data():
    # 提交剩余未批量的数据
    if hasattr(local_data, 'gathered_data') and local_data.gathered_data:
        print(f"Worker {multiprocessing.current_process().pid} 提交剩余结果: {local_data.gathered_data}")
        requests.post(POST_URL, json=local_data.gathered_data)
        local_data.gathered_data.clear()

def process_task(task):
    # 模拟任务执行，生成结果
    result = {"task_id": task, "result": f"done_{task}"}
    # 将结果加入当前Worker的缓存
    local_data.gathered_data.append(result)
    # 攒够批量大小就提交
    if len(local_data.gathered_data) >= BATCH_SIZE:
        print(f"Worker {multiprocessing.current_process().pid} 批量提交: {local_data.gathered_data}")
        requests.post(POST_URL, json=local_data.gathered_data)
        local_data.gathered_data.clear()
    return result

if __name__ == "__main__":
    # 创建进程池，指定初始化函数
    with multiprocessing.Pool(initializer=init_worker) as pool:
        # 处理百万级任务
        tasks = range(1000000)
        pool.map(process_task, tasks)

方案优势 #

• 每个Worker进程拥有独立的结果缓存，完全避免进程间数据竞争和共享问题。
• 进程退出回调不受任务执行状态影响，无论Worker正常完成任务还是异常终止，剩余结果都会被提交。
• 无需额外依赖全局队列或定时器，逻辑轻量化，适配百万级任务的高效执行场景。

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