标签: python-multiprocessing

我正在制作一个项目，从客户端传感器收集数据，处理收集到的数据并将其发送回客户端。可能有多个客户端要求同时从我们的服务器接收一些数据，所以我必须实现多重处理。我无法使用线程，因为某些变量必须与客户端无关。如果我这样做，我的代码可能会变得非常难以阅读和升级，而我不希望这样。所以我决定使用Processes，但是现在有一些数据需要在父进程和子进程之间进行剪切。经过一番研究，我发现 Pipe 通信可以满足我的要求。

以下代码成功地将数据从父进程发送到子进程，子进程更新数据并将其发送回父进程。但它之所以起作用只是因为 sleep() 函数阻止父级与子级同时使用管道。

如何更改它以使其执行相同的操作，但没有 sleep() 函数，我相信它很可能会在将来引起问题？

from multiprocessing import Process, Pipe
import time

def update_data(pipe):
    p_out, p_in = pipe
    L = []
    while True:
        message = p_out.recv()
        if message=='FINISHED':
            break       
        L.append(message)      

    L.append(['new data'])       #updating received data
    writer(L, p_in)              #sending received data to parent Process
    p_in.close()

def writer(i, p_in):
    p_in.send(i)
    p_in.send('FINISHED')

L = ['0' for i in range(10)]     #current data
if __name__=='__main__':
    p_out, p_in = Pipe()
    update_data_process = Process(target=update_data, args=((p_out, p_in),))
    update_data_process.start()    
    writer(L, p_in)              #sending current data to …

我正在尝试使用 python3 的多处理库的工作池来加速 CPU 密集型任务。我注意到，无论我的池中有多少工作人员，我总是会收到以下错误：

BrokenPipeError：[Errno 32] 管道损坏

导致此错误的代码如下：

def save_result(rincL):
    #Save results here

if __name__ == '__main__':
    p = Pool(8)
    allProcesses = []
    for i in range(60): 
        rincL = RINC_L(2, X_train, y_train[:,i].ravel(), X_test, y_test[:,i].ravel(), 6, i)
        allProcesses.append(p.apply_async(rincL.RINC_N, [0]))
    #Moved the two following lines based on suggestions in the comment
    #Still getting the same error.
    #p.close()
    #p.join()
    for process in allProcesses:
        save_result(process.get())
        time.sleep(1)
    p.close()
    p.join()

我还注意到该程序运行一段时间后。进程数量从 8 个跃升至一个非常高的数字。大约生成了 30 个进程：

我是多重处理的新手，如果有人可以在这里指导/帮助我，我将非常感激。我有以下 for 循环，它从两个函数获取一些数据。代码看起来像这样

    for a in accounts:
        dl_users[a['Email']] = get_dl_users(a['Email'], adConn)
        group_users[a['Email']] = get_group_users(a['Id'], adConn)

    print(f"Users part of DL - {dl_users}")
    print(f"Users part of groups - {group_users}")
    adConn.unbind()

这工作正常并获得所有结果，但最近我注意到它需要很多时间来获取用户列表，即 dl_users 和 group_users。大约需要 14-15 分钟才能完成。我正在寻找可以加速该函数的方法，并希望将此 for 循环转换为多处理。get_group_users并get_dl_users调用 LDAP，所以我不能 100% 确定是否应该将其转换为多处理或多线程。任何建议都会有很大帮助

多线程正在打印输出，但不是多处理。搜索堆栈溢出并回答问题并没有解决问题。

多重处理不起作用。

from threading import Thread
import datetime
from multiprocessing import Process
import sys


import time

def func1():

    print('Working')
    time.sleep(5)
    global a
    a=10
    print(datetime.datetime.now())


def func2():
    print("Working")
    time.sleep(10)
    print(datetime.datetime.now())
    

p1 = Process(target=func1)
p1.start()
p2 = Process(target=func2)
p2.start()
p1.join()
p2.join()


print(a)

即使 print(a) 也没有打印该值。它说

NameError: name 'a' is not defined

我正在使用Pipe模块Process（multiprocessingPython 3.8）。我的初始程序如下所示：

from multiprocessing import Process, Pipe


class Process1(object):
    def __init__(self, pipe_out):
        self.pipe_out = pipe_out

        self.run()

    def run(self):
        try:
            while True:
                print("Sending message to process 2")
                self.pipe_out.send(["hello"])
        except KeyboardInterrupt:
            pass


class Process2(object):
    def __init__(self, pipe_in):
        self.pipe_in = pipe_in

        self.run()

    def run(self):
        try:
            while self.pipe_in.poll():
                request = self.pipe_in.recv()
                method = request[0]
                args = request[1:]

                try:
                    getattr(self, method + "_callback")(*args)
                except AttributeError as ae:
                    print("Unknown callback received from pipe", str(ae))

            print("Process 2 done with receiving")
        except …