Python多进程程序如何统计每秒操作数？以批量图片下载任务为例

以下两种实现方式性能损耗极低，完全适配数千张图片的批量下载场景：

方案1：实时统计每秒下载次数（可动态查看当前速度） #

用multiprocessing.Value做进程安全的全局计数器，主进程单独开一个守护线程定时输出速度，方便你动态观察不同进程数下的速度变化：

import os
import requests
import time
from multiprocessing import Pool, Value
from threading import Thread

# 初始化进程共享计数器，'i'代表整数类型，初始值为0
complete_count = Value('i', 0)
start_time = None

# 子进程初始化函数，把共享计数器绑定到子进程全局变量
def init_counter(counter):
    global complete_count
    complete_count = counter

def download_image(image):
    save_dir = "[PATH TO SAVE IMAGES]"
    image_url = image['url']
    image_name = image['name']
    try:
        image_data = requests.get(image_url, timeout=10).content
        with open(os.path.join(save_dir, f"{image_name}.jpg"), 'wb') as f:
            f.write(image_data)
    finally:
        # 下载完成（不管成功失败）计数+1，自带锁保证计数安全
        with complete_count.get_lock():
            complete_count.value += 1

# 速度统计线程，每秒输出一次当前统计数据
def stat_speed(total_num):
    while True:
        elapsed = time.time() - start_time
        current_count = complete_count.value
        qps = current_count / elapsed if elapsed > 0 else 0
        # 可自定义输出格式，当前输出下载进度、平均QPS
        print(f"进度：{current_count}/{total_num} | 平均QPS：{qps:.2f} 张/秒", end='\r')
        if current_count >= total_num:
            break
        time.sleep(1)

if __name__ == "__main__":
    # 替换为你的图片列表
    images = [...]
    total = len(images)
    start_time = time.time()

    # 初始化进程池时传入计数器初始化方法和参数
    pool = Pool(8, initializer=init_counter, initargs=(complete_count,))
    
    # 启动速度统计线程
    stat_thread = Thread(target=stat_speed, args=(total,), daemon=True)
    stat_thread.start()

    # 执行下载任务
    pool.map(download_image, images)
    pool.close()
    pool.join()

    # 输出最终统计结果
    total_time = time.time() - start_time
    final_qps = total / total_time
    print(f"\n下载完成 | 总耗时：{total_time:.2f}s | 最终平均QPS：{final_qps:.2f} 张/秒")

小提示：如果需要只统计成功下载的QPS，把上述代码finally块里的计数逻辑，移动到write执行完成之后即可。

方案2：仅统计更精准的总平均QPS（不需要实时输出） #

如果不需要实时看速度，只是不想把进程池初始化、收尾的时间算进平均，可以用Pool.imap_unordered迭代已完成的任务，从第一个任务完成时开始计时，统计结果比直接算Pool全生命周期的平均更准确：

import os
import requests
import time
from multiprocessing import Pool

def download_image(image):
    save_dir = "[PATH TO SAVE IMAGES]"
    image_url = image['url']
    image_name = image['name']
    image_data = requests.get(image_url, timeout=10).content
    with open(os.path.join(save_dir, f"{image_name}.jpg"), 'wb') as f:
        f.write(image_data)
    return True

if __name__ == "__main__":
    # 替换为你的图片列表
    images = [...]
    total = len(images)
    pool = Pool(8)
    
    complete = 0
    start_time = None
    # 迭代已完成的任务，第一个任务返回时才开始计时，排除进程启动耗时
    for res in pool.imap_unordered(download_image, images):
        if start_time is None:
            start_time = time.time()
        complete += 1

    pool.close()
    pool.join()

    total_time = time.time() - start_time
    qps = complete / total_time
    print(f"平均QPS：{qps:.2f} 张/秒")

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