Python - Wie kann man mit mehreren Prozessen das Problem der langsamen Schleifenverschachtelung lösen?

Question

Es gibt ein Muster einer Schleife innerhalb einer Schleife.
Im Schleifenkörper der inneren Schleife müssen die Variablen der großen Schleife und der kleinen Schleife gleichzeitig verwendet werden.

Ich habe es hier zu einem einfachen Modell vereinfacht.
Dieses Modell wird sehr langsam sein, wenn die Funktion komplex ist.
Ich würde gerne fragen, wie man das Geschwindigkeitsproblem mit mehreren Prozessen lösen kann.

Meine Idee ist, Multiprozess nur für kleine Schleifen zu verwenden,
den Multiprozesscode in den Schleifenkörper der großen Schleife zu schreiben,
aber es schlägt immer wieder fehl,
Bitte Gott, den richtigen Code zu geben.

Danke!

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/...

之前测试过文中的例子，没啥问题，你沿着这些做的话应该是可以搞出来的