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

雨后青山
雨后青山
发布时间:2026-10-12 | 2人浏览过
原创

本文深入解析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)

总结

本篇文章就到这里了,希望能够给你带来帮助,也希望您能够多多关注本站的更多内容!

相关标签:

本站声明

本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。

Python 相关产品和应用

全部下载

专题列表