Python多进程通信性能优化：从Queue到Pipe的实战解析

Noah

发布时间：2026-10-16 | 7人浏览过

原创

本文针对Python多进程开发中Queue通信耗时过长的问题，通过视频帧处理的实际案例进行对比分析。演示了使用Queue时获取大数据的延迟现象，并提供了基于multiprocessing.Pipe结合线程预取内存队列的优化方案。实测数据显示，优化后数据获取间隔显著缩短且更加稳定，适用于对实时性要求较高的算法场景。

前言

近期在推进算法优化项目时，为了满足严格的实时性指标，必须将系统资源利用率挖掘到极致。为此，我们在算法的关键处理环节引入了多线程或多进程并行处理机制。

在代码调试阶段，发现进程间使用Queue进行数据传递时耗时显著增加，尤其是在传输图片等大容量数据时，这一瓶颈尤为明显。

读者可先参考下方的Demo代码，判断其是否契合您当前的业务场景。

随后将介绍具体的解决方案。

使用进程间Queue效率问题场景

以下代码示例模拟了从两个视频源读取图像帧并进行处理的过程。

#!/user/bin/env python
# coding=utf-8
"""
@project : csdn-pro
@author  : 剑客阿良_ALiang
@file   : test13.py
@ide    : PyCharm
@time   : 2022-09-13 10:47:35
"""
import time
 
import cv2
from multiprocessing import Queue, Process
 
 
def fun1(q: Queue):
    cap = cv2.VideoCapture("11.mp4")
    a = []
    while cap.isOpened():
        ret, frame = cap.read()
        if ret:
            a.append(frame)
            if len(a) == 25:
                q.put(a)
                a = []
            time.sleep(0.038)
 
 
def fun2(q: Queue):
    cap = cv2.VideoCapture("3333333.mp4")
    a = []
    while cap.isOpened():
        ret, frame = cap.read()
        if ret:
            a.append(frame)
            if len(a) == 25:
                q.put(a)
                a = []
            time.sleep(0.038)
 
 
def fun3(q1: Queue, q2: Queue, q3: Queue):
    while True:
        st0 = time.time()
        a1 = q1.get()
        st1 = time.time()
        a2 = q2.get()
        st2 = time.time()
        print("{} 耗时:{} - {}".format(time.time(), st1 - st0, st2 - st1))
        q3.put((a1, a2))
 
 
def fun4(q3: Queue):
    while True:
        st0 = time.time()
        a1, a2 = q3.get()
        et = time.time()
        print("hhhh耗时: {}".format(et - st0))
 
 
if __name__ == '__main__':
    q1 = Queue()
    q2 = Queue()
    q3 = Queue()
    p1 = Process(target=fun1, args=(q1,))
    p2 = Process(target=fun2, args=(q2,))
    p3 = Process(target=fun3, args=(q1, q2, q3,))
    p4 = Process(target=fun4, args=(q3,))
    p1.start()
    p2.start()
    p3.start()
    p4.start()
    p1.join()
    p2.join()
    p3.join()
    p4.join()

代码逻辑说明：

1、上述逻辑模拟每秒读取25帧图像，并将包含25帧图像的列表一次性放入队列中。

接下来观察从Queue获取图像列表的效率。部分执行日志如下所示。

1663139091.3648114 耗时:1.6036181449890137 - 0.1361703872680664 hhhh耗时: 3.0635826587677 1663139093.056612 耗时:1.5302414894104004 - 0.1615591049194336 hhhh耗时: 1.6867034435272217 1663139094.7388775 耗时:1.5256507396697998 - 0.1566147804260254 hhhh耗时: 1.6849782466888428 1663139096.36547 耗时:1.4680161476135254 - 0.15857625007629395 hhhh耗时: 1.651228427886963 1663139097.9867501 耗时:1.4417593479156494 - 0.179520845413208 hhhh耗时: 1.609663963317871 1663139099.5894623 耗时:1.4391484260559082 - 0.16356372833251953 hhhh耗时: 1.7086796760559082 1663139101.3031366 耗时:1.5481102466583252 - 0.16556406021118164 hhhh耗时: 1.657604455947876 1663139102.9448056 耗时:1.470097303390503 - 0.1715717315673828 hhhh耗时: 1.5316739082336426 1663139104.5233243 耗时:1.4139580726623535 - 0.16456055641174316

Process finished with exit code -1

从日志可以看出，从进程队列中获取数据的耗时较长。以q3中同时获取的时间（蓝色标记部分）为例，其数值远大于1秒。

而整体获取图像帧的效率（红色标记部分）显示，时间间隔也大于1秒，无法满足高实时性需求。

采用管道模式解决

鉴于上述时间间隔不可接受，此处改用multiprocessing.Pipe管道机制来提前输入图片数据。

优化后的样例代码如下：

#!/user/bin/env python
# coding=utf-8
"""
@project : csdn-pro
@author  : 剑客阿良_ALiang
@file   : test13.py
@ide    : PyCharm
@time   : 2022-09-13 10:47:35
"""
import threading
import time
 
import cv2
from multiprocessing import Queue, Process, Pipe
 
 
def fun1(pipe_in):
    cap = cv2.VideoCapture("11.mp4")
    while cap.isOpened():
        ret, frame = cap.read()
        if ret:
            ret, frame = cap.read()
            pipe_in.send((int(time.time()), frame))
            time.sleep(0.038)
 
 
def fun2(pipe_in):
    cap = cv2.VideoCapture("3333333.mp4")
    while cap.isOpened():
        ret, frame = cap.read()
        if ret:
            ret, frame = cap.read()
            pipe_in.send((int(time.time()), frame))
            time.sleep(0.038)
 
 
def fun3(pipe_rev1, pipe_rev2):
    def handle(pipe_rev1, q1):
        _cul = 0
        a = []
        while True:
            _t, _frame = pipe_rev1.recv()
            if _cul == 0:
                a.append(_frame)
                _cul = _t
            elif _t > _cul != 0:
                if len(a) != 0:
                    q1.put(a)
                    _cul = _t
                    a = []
                    a.append(_frame)
            elif _t == _cul != 0:
                a.append(_frame)
 
    q1 = Queue()
    q2 = Queue()
    threading.Thread(target=handle, args=(pipe_rev1, q1,)).start()
    threading.Thread(target=handle, args=(pipe_rev2, q2,)).start()
    while True:
        if not q1.empty() and not q2.empty():
            st0 = time.time()
            _f1 = q1.get()
            st1 = time.time()
            _f2 = q2.get()
            et = time.time()
            print("{} 耗时：{} - {}".format(time.time(), st1 - st0, et - st1))
 
 
if __name__ == '__main__':
    pipe_in1, pipe_out1 = Pipe()
    pipe_in2, pipe_out2 = Pipe()
    p1 = Process(target=fun1, args=(pipe_in1,))
    p2 = Process(target=fun2, args=(pipe_in2,))
    p3 = Process(target=fun3, args=(pipe_out1, pipe_out2,))
    p1.start()
    p2.start()
    p3.start()
 
    p1.join()
    p2.join()
    p3.join()

代码逻辑说明：

1、通过两个线程持续从管道接收数据，并将其写入内存中的Queue，从而将数据提前加载到当前进程的内存中。

观察优化后的时间间隔是否趋于稳定，部分执行日志如下：

1663139886.0722673 耗时：0.003930091857910156 - 0.005983591079711914 1663139887.6837587 耗时：0.09677457809448242 - 0.09172177314758301 1663139888.472634 耗时：0.061833858489990234 - 0.05984067916870117 1663139889.5441313 耗时：0.07132482528686523 - 0.07080578804016113 1663139890.548978 耗时：0.06183457374572754 - 0.06881546974182129 1663139891.5112402 耗时：0.0637204647064209 - 0.0718080997467041 1663139892.4756596 耗时：0.06682205200195312 - 0.06978344917297363 1663139893.5788367 耗时：0.06779074668884277 - 0.07928323745727539

可以看出，时间间隔变得较为稳定。