Python多进程编程详解：实现、通信与进程池

雨后青山

发布时间：2026-10-12| 5人浏览过

原创

本文深入解析Python多进程编程，涵盖多进程创建、优缺点分析。重点讲解Queue和Pipe两种进程间通信机制及Manager共享数据方法，并详细介绍进程池Pool的使用，包括异步执行、同步执行及资源管理，旨在提升多核环境下的程序执行效率。

一、多进程的实现

方法一

# 方法包装   多进程
from multiprocessing import Process
from time import sleep
def func1(arg):
    print(f'{arg}开始...')
    sleep(2)
    print(f'{arg}结束...')
if __name__ == "__main__":
    p1 = Process(target=func1,args=('p1',))
    p2 = Process(target=func1,args=('p2',))
    p1.start()
    p2.start()

方法二：

二、使用进程的优缺点

1、优点

充分利用计算机的多核CPU资源，实现任务的并发执行，从而显著提升程序的运行效率。
各进程运行相互独立，互不干扰，且创建过程相对简便。
进程间拥有独立的空间，数据安全性更高。

2、缺点

进程的创建和销毁会消耗较多的系统资源。

三、进程的通信

Python提供了多种机制来实现进程间通信（IPC），主要包含以下两种：

1. Python multiprocessing 模块下的 Queue 类，提供了多种方法以支持多个进程间的通信。

2. Pipe（管道），通常用于实现两个进程之间的通信，这两个进程分别位于管道的两端。

Pipe直译为“管”或“管道”，这种多进程编程方式与实际生活中的管道非常相似。通常情况下，管道有两个端口，而Pipe也常用于实现两个进程之间的通信，这两个进程分别位于管道的两端，一端负责发送数据，另一端负责接收数据 - send(obj)

向管道的另一端发送一个对象obj，接收端使用 recv() 方法进行接收。需要注意的是，该对象必须是可序列化的；如果序列化后的对象大小超过 32MB，则极有可能引发 ValueError 异常 - recv()

接收另一端通过 send() 方法发送过来的数据 - close()

关闭连接 - poll([timeout])

检查并返回连接中是否还有数据可以读取 - end_bytes(buffer[, offset[, size]])

发送字节数据。若未指定 offset 和 size 参数，则默认发送 buffer 字节串的全部数据；若指定了这两个参数，则仅发送从 offset 开始、长度为 size 的字节数据。通过该方法发送的数据，应当使用 recv_bytes() 或 recv_bytes_into 方法进行接收 - recv_bytes([maxlength])

接收通过 send_bytes() 方法发送的数据，maxlength 用于指定最多接收的字节数。该方法返回接收到的字节数据 - recv_bytes_into(buffer[, offset])

功能与 recv_bytes() 方法类似，区别在于该方法将接收到的数据直接放入 buffer 中。

1、Queue 实现进程间通信

from multiprocessing import Process,current_process,Queue   # current_process 指的是当前进程
# from queue import Queue
import os
def func(name,mq):
    print('进程ID {} 获取了数据：{}'.format(os.getpid(),mq.get()))
    mq.put('shiyi')
if __name__ == "__main__":
    # print('进程ID:{}'.format(current_process().pid))
    # print('进程ID:{}'.format(os.getpid()))
    mq = Queue()
    mq.put('yangyang')
    p1 = Process(target=func,args=('p1',mq))
    p1.start()
    p1.join()
    print(mq.get())

2、Pipe 实现进程间通信（一边发送send(obj),一边接收(obj)）

from multiprocessing import Process,current_process,Pipe
import os
def func(name,con):
    print('进程ID {} 获取了数据：{}'.format(os.getpid(),con.recv()))
    con.send('你好！')
if __name__ == "__main__":
    # print('进程ID:{}'.format(current_process().pid))
    con1,con2 = Pipe()
    p1 = Process(target=func,args=('p1',con1))
    p1.start()
    con2.send("hello!")
    p1.join()
    print(con2.recv())

四、Manager管理器

管理器提供了一种创建共享数据的方法，从而可以在不同进程中共享数据。

from multiprocessing import Process,current_process
import os
from multiprocessing import Manager
def func(name,m_list,m_dict):
    print('子进程ID {} 获取了数据：{}'.format(os.getpid(),m_list))
    print('子进程ID {} 获取了数据：{}'.format(os.getpid(),m_dict))
    m_list.append('你好')
    m_dict['name'] = 'shiyi'    
if __name__ == "__main__":
    print('主进程ID:{}'.format(current_process().pid))
    with Manager() as mgr:
        m_list = mgr.list()
        m_dict = mgr.dict()
        m_list.append('Hello!!')
        p1 = Process(target=func,args=('p1',m_list,m_dict))
        p1.start()
        p1.join()
        print(m_list)
        print(m_dict)

五、进程池

Python 提供了更高效的多个进程管理方式，即使用进程池。

进程池可以向用户提供指定数量的进程。当有新的请求提交到进程池中时，如果池未满，则会创建一个新的进程来执行该请求；反之，如果池中的进程数已达到规定的最大值，那么该请求就会进入等待状态，只要池中有进程空闲下来，该请求就能得到执行。

使用进程池的优点

1. 提高效率，节省开辟进程和开辟内存空间的时间及销毁进程的时间

2. 节省内存空间

类/方法	功能	参数
Pool(processes)	创建进程池对象	processes 表示进程池中有多少进程
pool.apply_async(func,a rgs,kwds)	异步执行；将事件放入到进程池队列	func 事件函数 args 以元组形式给 func 传参 kwds 以字典形式给 func 传参返回值：返回一个代表进程池事件的对象，通过返回值的 get 方法可以得到事件函数的返回值
pool.apply(func,args,kw ds)	同步执行；将事件放入到进程池队列	func 事件函数 args 以元组形式给 func 传参 kwds 以字典形式给 func 传参
pool.close()	关闭进程池
pool.join()	回收进程池
pool.map(func,iter)	类似于 python 的 map 函数，将要做的事件放入进程池	func 要执行的函数 iter 迭代对象

from multiprocessing import Pool
import os
from time import sleep
def func1(name):
    print(f"当前进程的ID:{os.getpid()},{name}")
    sleep(2)
    return name
def func2(args):
    print(args)
if __name__ == "__main__":
    pool = Pool(5)
    pool.apply_async(func = func1,args=('t1',),callback=func2)
    pool.apply_async(func = func1,args=('t2',),callback=func2)
    pool.apply_async(func = func1,args=('t3',),callback=func2)
    pool.apply_async(func = func1,args=('t4',))
    pool.apply_async(func = func1,args=('t5',))
    pool.apply_async(func = func1,args=('t6',))
    pool.close()
    pool.join()

from multiprocessing import Pool
import os
from time import sleep
def func1(name):
    print(f"当前进程的ID:{os.getpid()},{name}")
    sleep(2)
    return name
if __name__ == "__main__":
   with Pool(5) as pool:
        args = pool.map(func1,('t1,','t2,','t3,','t4,','t5,','t6,','t7,','t8,'))
        for a in args:
            print(a)