python - 如何利用多進程來解決循環嵌套速度不行的問題？

Question

有一個循環裡面套循環的模式，
在內循環的循環體內要同時用到大循環和小循環的變數。

我這裡是簡化成了一個簡單的模型，
這種模式如果函數複雜的話速度超級慢，
想問一下如何使用多進程的辦法來解決速度問題？

我的思路是，只對小循環採用多進程，
在大循環的循環體內寫多進程的程式碼，
但是一直失敗，
求大神給出正確的程式碼。

拜謝！

import random as r
list1=list(range(100))
i=0
reslist=[]
while i<2000:#大循环
    alist=[]#三个列表变量，每次循环开始时清空
    blist=[]
    clist=[]
    for each in list1:#小循环
        x=r.randint(i+30,i+60)+each#涉及到大、小循环变量的几个函数，这里用random示意
        y=r.randint(i+60,i+120)+each
        z=r.randint(i+60,i+180)+each
        
        res=2.5*x-y-z
        reslist.append(res)#对函数结果进行操作
        if res>=50:
            alist.append(each)
        if -50<res<50:
            blist.append(each)
        if res<=-50:
            clist.append(each)
            
    for each in alist:#在大循环中对小循环中得出的结果进行进一步其他操作
        print(each)
    for each in blist:
        print(each)
    for each in clist:
        print(each)
    
    i+=1

Answer 1

首先，平行計算需要各個平行運算的子程式間沒有相互因果關係。
小循環內，res與x,y,z，與alist,blist,clist，都是因果關係密切的，很難拆分並行計算。
題主貼上來的雖然不是原始程式碼，不知道原始程式碼裡大循環間有沒有因果關係，不過從示意程式碼來看，
把大循環拆分為N個執行緒（用不到進程吧）應該是可以的，每個執行緒計算2000/N次。
例如，分成8個線程，線程1計算i=0到249，線程2計算i=250到499，依序類推。 。 。
這裡N的大小，可以根據CPU的核數來定，如果N超過CPU的核數，就沒有太大意義了，反而有可能會降低效率。

Answer 2

中間要用elif吧，最後面for的縮排好像也有問題

Answer 3

可以在大循環這裡開多進程，例如大循環2000次，如CPU的核數是4，則開4個進程，每個進程負責運行500個

小循環結束後，可以開子執行緒執行下面的這些後續操作，大循環繼續往前處理

for each in alist:#在大循环中对小循环中得出的结果进行进一步其他操作
    print(each)
for each in blist:
    print(each)
for each in clist:
    print(each)

Answer 4

可以將小循環用子進程去處理 不過這樣 你需要兩個大循環。一個循環處理小循環 ，等處理完這個循環在來個大循環處理後面的事情

像這樣

import random as r


def cumput(i, list1):
    alist = []
    blist = []
    clist = []
    reslist = []
    for each in list1:  # 小循环
        x = r.randint(i + 30, i + 60) + each  # 涉及到大、小循环变量的几个函数，这里用random示意
        y = r.randint(i + 60, i + 120) + each
        z = r.randint(i + 60, i + 180) + each

        res = 2.5 * x - y - z
        reslist.append(res)  # 对函数结果进行操作
        if res >= 50:
            alist.append(each)
        if -50 < res < 50:
            blist.append(each)
        if res <= -50:
            clist.append(each)
    return alist, blist, clist, reslist


if __name__ == '__main__':
    multiprocessing.freeze_support()
    list1 = list(range(100))
    i = 0
    pool = multiprocessing.Pool(2)
    res = {}
    while i < 2000:  # 大循环
        res[i]=pool.apply_async(cumput, (i, list1,))
        i += 1
    pool.close()
    pool.join()
    for i in res:
        for each in res[i].get()[0]:  # 在大循环中对小循环中得出的结果进行进一步其他操作
            print(each)
        for each in res[i].get()[1]:
            print(each)
        for each in res[i].get()[2]:
            print(each)

Answer 5

如果小循環中執行的函數比較耗時的話可以考慮生產者-消費者模型

import random
from threading import Thread
from Queue import Queue

resqueue = Queue()
aqueue = Queue()
bqueue = Queue()
cqueue = Queue()

def producer():
    list1=list(range(100))
    
    for _ in range(2000):
        for each in list1:
            x=r.randint(i+30,i+60)+each
            y=r.randint(i+60,i+120)+each
            z=r.randint(i+60,i+180)+each
            
            res=2.5*x-y-z
            resqueue.put(res)
            
            if res>=50:
                aqueue.put(each)
            if -50<res<50:
                bqueue.put(each)
            if res<=-50:
                cqueue.put(each)

def consumer_a():
    while True:
        try:
            data = aqueue.get(timeout=5)
        except Queue.Empty:
            return
        else:
            # 耗时操作
            deal_data(data)
            aqueue.task_done()
            
def consumer_b():
    while True:
        try:
            data = bqueue.get(timeout=5)
        except Queue.Empty:
            return
        else:
            # 耗时操作
            deal_data(data)
            bqueue.task_done()
            
 def consumer_c():
    while True:
        try:
            data = cqueue.get(timeout=5)
        except Queue.Empty:
            return
        else:
            # 耗时操作
            deal_data(data)
            cqueue.task_done()

 def consumer_res():
    while True:
        try:
            data = resqueue.get(timeout=5)
        except Queue.Empty:
            return
        else:
            # 耗时操作
            deal_data(data)
            resqueue.task_done()
            
if __name__ == "__main__":
    t1 = Thread(target=producer)
    t2 = Thread(target=consumer_a)
    ...
    
    t1.start()
    t2.start()
                       

Answer 6

題主是不是應該先設計好進程的輸入與輸出，多進程做並行計算的話進程之間的通信是最重要的，據我了解的應該是MPI，比如多層循環，應該是先分發部分數據到每個進程，每個進程做計算後再回到資料整合點，然後合併結果輸出。

還有一個比較重要的點是估算每個行程的執行時間，畢竟有進程間的通訊的話等待時間也會導致效率下降。

@一代鍵客 所說，你的嵌套不太符合平行計算的輸入規則，可以看看這個例子

http://blog.csdn.net/zouxy09/...

之前測試過文中的例子，沒啥問題，你沿著這些做的話應該是可以搞出來的