调用类方法时Python Multiprocessing无法并发的疑问及原因探究

问题概述 #

测试Python multiprocessing机制时发现两个现象：

• 用Process启动类对象时，接收方能在发送方完成前打印数据，实现了并发
• 用Process调用类方法时，接收方必须等发送方完全执行完才打印数据，无法并发（采用Pipe通信）

核心疑问 #

1. 为何调用类方法无法实现并发？
2. 类/函数调用能正常并发的原因是什么？二者差异在哪？

问题解答 #

1. 调用类方法无法并发的原因 #

你在启动发送方进程时犯了一个关键错误：

sender_process = Process(target=app.worker(), args=())

这里的app.worker()是直接在主进程中执行了方法，而非将方法对象作为可调用目标传给Process。这会导致worker()在主进程里同步跑完所有逻辑后，才会创建子进程——此时子进程没有任何任务可执行，自然看不到接收方和发送方的并发效果。

正确写法应该是传递方法对象本身（不要加括号）：

sender_process = Process(target=app.worker, args=())

2. 类/函数调用能正常并发的原因及差异 #

在第二个示例中，你将类sender直接传给了Process的target参数：

sender_process = Process(target=sender, args=(1,2,conn2,))

multiprocessing.Process的target接受任何可调用对象（函数、类、实现__call__的对象都可以）。当target是类时，子进程会独立执行该类的构造方法__init__，而你的__init__里包含了发送数据的核心逻辑，因此子进程启动后会和主进程、接收方进程并行执行，实现真正的并发。

二者核心差异：

• 错误示例：提前在主进程执行了方法逻辑，子进程无任务可跑，本质是单进程同步执行
• 正确示例：将可调用对象（类/函数）传给target，子进程在独立的进程空间执行逻辑，实现进程隔离与并发

错误示例（调用类方法） #

from multiprocessing import Process, Pipe
from time import sleep

class inherit_1:
    def __init__(self, a):
        self.a = a

class inherit_2:
    def __init__(self, b):
        self.b = b

class sender(inherit_1, inherit_2):
    def __init__(self, a, b, conn2): 
        inherit_1.__init__(self, a)
        inherit_2.__init__(self, b) 
        self.conn2 = conn2
          
    # generate work
    def worker(self):    
        print('Sender: Running', flush=True)
        for i in range(10):
            # generate a value
            value = i*self.a + i*self.b
            sleep(1)
            self.conn2.send(value) 
        # all done
        self.conn2.send(None)
        print('Sender: Done', flush=True)

class receiver:
    def __init__(self, conn1):
        self.conn1 = conn1
        print('Receiver: Running', flush=True)
        while True:
            # get a unit of work
            item = conn1.recv() 
            print(f'>receiver got {item}', flush=True)
            # check for stop
            if item is None:
                break
        # all done
        print('Receiver: Done', flush=True)

if __name__ == '__main__': 
    
    # create the pipe
    conn1, conn2 = Pipe()

    # Create an Instance of the class
    app = sender(1,2,conn2)  

    # 错误：直接调用worker()，导致主进程同步执行完逻辑才启动子进程
    sender_process = Process(target=app.worker(), args=()) 
    sender_process.start()

    # start the receiver
    receiver_process = Process(target=receiver, args=(conn1,))
    receiver_process.start()

    sender_process.join()
    receiver_process.join()

正确示例（启动类对象） #

from multiprocessing import Process, Pipe
from time import sleep

class inherit_1:
    def __init__(self, a):
        self.a = a

class inherit_2:
    def __init__(self, b):
        self.b = b

class sender(inherit_1, inherit_2):
    def __init__(self, a, b, conn2): 
        self.conn2 = conn2
        self.a = a
        self.b = b
        # generate work  
        print('Sender: Running', flush=True)
        for i in range(10):
            # generate a value
            value = i*self.a + i*self.b
            # block
            sleep(1)
            # send data
            conn2.send(value)
        # all done
        conn2.send(None)
        print('Sender: Done', flush=True) 

class receiver: 
    def __init__(self, conn1):
        self.conn1 = conn1
        print('Receiver: Running', flush=True)
        while True:
            # get a unit of work
            item = conn1.recv() 
            # report
            print(f'>receiver got {item}', flush=True)
            # check for stop
            if item is None:
                break
        # all done
        print('Receiver: Done', flush=True)

if __name__ == '__main__': 

    # create the pipe
    conn1, conn2 = Pipe()

    # 正确：将类作为可调用对象传给target，子进程独立执行__init__逻辑
    sender_process = Process(target=sender, args=(1,2,conn2,))
    sender_process.start()

    # start the receiver
    receiver_process = Process(target=receiver, args=(conn1,)) 
    receiver_process.start()

    sender_process.join()
    receiver_process.join()

内容的提问来源于stack exchange，提问作者Rambo