Python解決多進(jìn)程間訪問(wèn)效率低的方法總結(jié)

更新時(shí)間：2022年09月14日 16:00:53 作者：劍客阿良_ALiang

這篇文章主要為大家詳細(xì)介紹了當(dāng)Python多進(jìn)程間訪問(wèn)效率低時(shí)，應(yīng)該如何解決？文中的示例代碼講解詳細(xì)，感興趣的小伙伴可以了解一下

前言

最近在解決一些算法優(yōu)化的問(wèn)題，為了實(shí)時(shí)性要求，必須精益求精的將資源利用率用到極致。同時(shí)對(duì)算法中一些處理進(jìn)行多線程或者多進(jìn)程處理。

在對(duì)代碼的調(diào)試過(guò)程中，發(fā)現(xiàn)在進(jìn)程間隊(duì)列使用耗時(shí)很長(zhǎng)，特別是圖片這種比較大的數(shù)據(jù)的時(shí)候。

可以先看一下我下面的demo是不是符合你的場(chǎng)景。

下面還有我的解決方案。

使用進(jìn)程間Queue效率問(wèn)題場(chǎng)景

代碼樣例如下，模擬從兩個(gè)視頻讀取圖片幀進(jìn)行處理。

#!/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("{} 耗時(shí):{} - {}".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耗時(shí): {}".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()

代碼說(shuō)明：

1、上面模擬每秒25幀讀取圖片，并傳遞一個(gè)25幀的圖片list給到隊(duì)列。

我們看一下從queue獲取圖片list的效率。部分執(zhí)行結(jié)果如下。

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

Process finished with exit code -1

可以看出我們從進(jìn)程隊(duì)列g(shù)et數(shù)據(jù)的耗時(shí)很長(zhǎng)，從q3中同時(shí)獲取的時(shí)間如藍(lán)色標(biāo)記，遠(yuǎn)大于1秒鐘。

而整體獲取圖片幀的效率如紅色標(biāo)記，間隔時(shí)間大于1秒。

采用管道模式解決

這個(gè)時(shí)間間隔沒(méi)法接受，我才用multiprocessing.Pipe管道來(lái)提前輸入圖片。

樣例代碼如下：

#!/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("{} 耗時(shí)：{} - {}".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()

代碼說(shuō)明：

1、通過(guò)兩個(gè)線程不停從管道接受并寫(xiě)到內(nèi)存的Queue里面，提前放到當(dāng)前進(jìn)程內(nèi)存里。

看一下間隔是否穩(wěn)定，部分執(zhí)行結(jié)果如下

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

時(shí)間間隔還是比較穩(wěn)定的。

總結(jié)

如果你遇到和我一樣的場(chǎng)景，可以仔細(xì)觀察一下進(jìn)程間數(shù)據(jù)是否傳輸?shù)谋容^慢?？梢钥紤]和我一樣的方式來(lái)解決。

到此這篇關(guān)于Python解決多進(jìn)程間訪問(wèn)效率低的方法總結(jié)的文章就介紹到這了,更多相關(guān)Python多進(jìn)程訪問(wèn)效率低內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: