关于python多进程中的常用方法详解
更新时间：2023年07月26日 09:03:45 作者：IT之一小佬
这篇文章主要介绍了关于python多进程中的常用方法详解,python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU资源，在python中大部分情况需要使用多进程,需要的朋友可以参考下
1、进程间的通信
全局变量在多个进程中不共享资源，进程之间的数据是独立的，默认情况下是互不影响的。
示例代码：
from multiprocessing import Process
num = 1
def task1():
global num
num += 5
print("子进程1运行，num:", num)
def task2():
global num
num += 10
print("子进程2运行，num:", num)
if __name__ == '__main__':
print("父进程开始运行...")
p1 = Process(target=task1)
p2 = Process(target=task2)
p1.start()
p2.start()
p1.join()
p2.join()运行结果：
2、用 Queue()实现多进程之间的数据传递
Queue 是多进程安全的队列，可以使用 Queue 实现多进程之间的数据传递。
put 方法用以插入数据到队列中， put 方法还有两个可选参数： blocked 和 timeout。如果 blocked 为 True（默认值），并且 timeout 为正值，该方法会阻塞 timeout 指定的时间，直到该队列有剩余的空间。如果超时，会抛出 Queue.full 异常。如果 blocked 为 False，但该 Queue 已满，会立即抛出 Queue.full 异常。
get 方法可以从队列读取并且删除一个元素。同样， get 方法有两个可选参数： blocked和 timeout。如果 blocked 为 True（默认值），并且 timeout 为正值，那么在等待时间内没有取到任何元素，会抛出 Queue.Empty 异常。如果 blocked 为 False，有两种情况存在，如果Queue 有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty 异常
示例代码：
from multiprocessing import Queue
q = Queue(3)
q.put('msg1')
q.put('msg2')
print('消息队列是否已满：', q.full())
q.put('msg3')
print('消息队列是否已满：', q.full())
# q.put('msg4') # 以为消息队列已经满了，需要直接写入需要等待，如果超时会抛出异常
# 写入数据时先判断，判断队列是否已满
if not q.full():
q.put('msg4')
# 同理，取消息时可以先判断队列是否有数据
if not q.empty():
for _ in range(q.qsize()):
print(q.get())运行结果：
示例代码：
from multiprocessing import Process
from multiprocessing import Queue
import random
import os
# 向queue中输入数据的函数
def inputQ(queue):
info = random.randint(1, 100)
queue.put(info)
print('进程{}往队列中存了一个数据：{}'.format(os.getpid(), info))
# 向queue中输出数据的函数
def outputQ(queue):
info = queue.get()
print('进程{}从队列中取出一个数据：{}'.format(os.getpid(), info))
if __name__ == '__main__':
queue = Queue(5)
lst_1 = []
lst_2 = []
for i in range(3):
process = Process(target=inputQ, args=(queue,))
process.start()
lst_1.append(process)
# 输出进程
for i in range(2):
process = Process(target=outputQ, args=(queue,))
process.start()
lst_2.append(process)
for p in lst_1:
p.join()
for p in lst_2:
p.join()运行结果：
3、Queue 队列实现进程间通信
示例代码：
import time
from multiprocessing import Queue, Process
def write_data(q):
# 将列表元素写入到队列中
for i in ['aa', 'bb', 'cc', 'dd']:
print('开始写入值%s' %i)
q.put(i)
time.sleep(1)
def read_data(q):
print("开始读取数据...")
while True:
if not q.empty():
print("读取到数据：", q.get())
time.sleep(1)
else:
break
if __name__ == '__main__':
# 创建队列
q = Queue()
# 创建进程
qw = Process(target=write_data, args=(q, ))
qr = Process(target=read_data, args=(q, ))
# 启动进程
qw.start()
qr.start()
qw.join()
qr.join()运行结果：
4、Manage()的使用
如果使用 Pool 创建进程，就需要使用 multiprocessing.Manager()中的 Queue()来完成进程间的通信，而不是 multiprocessing.Queue()，否则会抛出异常。
示例代码：
import time
from multiprocessing import Manager, Pool, Queue
def write_data(q):
# 将列表元素写入到队列中
for i in ['aa', 'bb', 'cc', 'dd']:
print('开始写入值%s' %i)
q.put(i)
time.sleep(1)
def read_data(q):
print("开始读取数据...")
while True:
if not q.empty():
print("读取到数据：", q.get())
time.sleep(1)
else:
break
if __name__ == '__main__':
# 创建队列
q = Manager().Queue()
# q = Queue() # 直接这样使用Queue()会报错
# 创建进程池
p = Pool(3)
# 使用apply阻塞模式创建进程
p.apply(write_data, (q, ))
p.apply(read_data, (q, ))
p.close()
p.join()运行结果：
示例代码：
import os
from multiprocessing import Manager, Process
# 定义了一个foo函数,接收一个字典和一个列表
def foo(dic, lst):
# 字典和列表都放进程ID
dic[os.getpid()] = os.getpid()
lst.append(os.getpid())
if __name__ == '__main__':
# 生成Manager对象
manager = Manager()
dic = manager.dict()
print(dic)
lst = manager.list(range(3))
print(lst)
# 10个进程分别join
p_list = []
for i in range(10):
p = Process(target=foo, args=(dic, lst))
p.start()
p_list.append(p)
for res in p_list:
res.join()
# 打印字典和列表
print(dic)
print(lst)运行结果：
示例代码：
from multiprocessing import Pool, current_process, Manager
import time
def produce_data(queue):
for i in range(10):
queue.put(i)
def consume_data(queue):
while queue.qsize() > 0:
data = queue.get() # 注意：当get()拿不到数据时，会一直处于等待状态
print(f"当前进程为：{current_process().name}, 队列获取数据为：{data},队列剩余数据为：{queue.qsize()}个！")
time.sleep(0.01)
if __name__ == '__main__':
print(f"主进程{current_process().name}开始执行！")
p = Pool(processes=6, maxtasksperchild=6)
queue = Manager().Queue(maxsize=20)
p.apply_async(produce_data, args=(queue, ))
time.sleep(1)
for i in range(5):
p.apply_async(consume_data, args=(queue, ))
p.close() # 关闭进程池，防止将任何其他任务提交到池中。需要在join之前调用，否则会报ValueError: Pool is still running错误
p.join() # 等待进程池中的所有进程执行完毕
print(f"主进程{current_process().name}结束！")运行结果：
注意：
- p.close() # 关闭进程池，防止将任何其他任务提交到池中。需要在join之前调用，否则会报ValueError: Pool is still running错误
- p.join() # 等待进程池中的所有进程执行完毕
- p.close()是关掉进程池子，是不再向里面添加进程了，对Pool对象调用join()方法会等待所有子进程执行完毕，调用join()之前必须先调用close()，调用close()之后就不能继续添加新的Process了。
5、current_process()的使用
示例代码：
from multiprocessing import Process, current_process
import time
import random
lst = []
def task(i):
print(current_process().name, i, 'start...') # current_process().name输出进程的名字
time.sleep(random.randint(1, 4))
lst.append(i)
print(lst)
print(current_process().name, i, 'end.....')
if __name__ == "__main__":
p_lst = []
for i in range(4):
p = Process(target=task, args=(i, ))
p_lst.append(p)
p.start()
for p in p_lst:
p.join() # 阻塞当前进程，直到子进程全部退出
print("main end.......")运行结果：
6、进程池
进程池里有固定数量的进程，每次执行任务时都从进程池中取出一个空闲进程来执行，如果任务数量超过进程池中进程数量，那么就等待已经在执行的任务结束之后，有进程空闲之后再执行，也就是说，同一时间，只有固定数量的进程在执行，这样对操作系统得压力也不会太大，效率也得到保证。
示例代码：
from multiprocessing import Pool, current_process
import time
import random
lst = []
def task(i):
print(current_process().name, i, 'start...')
time.sleep(random.randint(1, 5))
lst.append(i)
print(lst)
print(current_process().name, i, 'end.....')
if __name__ == "__main__":
p = Pool(processes=3, maxtasksperchild=3)
for i in range(10):
p.apply_async(func=task, args=(i,)) # 进程池接收任务
p.close() # 关闭进程池 ==》 不接受任务
p.join() # 等待子进程执行完毕，父进程再执行
print("end.............")运行结果：
示例代码： 【同步执行】
import os
import time
import random
from multiprocessing import Pool
def func1(n):
print('任务{}开始执行，进程为：{}'.format(n, os.getpid()))
time.sleep(random.randint(1, 4))
print('任务{}结束执行，进程为：{}'.format(n, os.getpid()))
if __name__ == '__main__':
# c创建一个进程池，里面有三个进程
p = Pool(3)
for i in range(10):
res = p.apply(func1, args=(i,))运行结果：
示例代码： 【异步执行】
import os
import time
import random
from multiprocessing import Pool
def func1(n):
print('任务{}开始执行，进程为：{}'.format(n, os.getpid()))
time.sleep(random.randint(1, 4))
print('任务{}结束执行，进程为：{}'.format(n, os.getpid()))
if __name__ == '__main__':
# c创建一个进程池，里面有三个进程
p = Pool(3)
for i in range(5):
res = p.apply_async(func1, args=(i,))
p.close() # 一定要关闭
p.join() # 一定要使用join，不然进程池里的进程没来得及执行，主进程结束了，子进程也都跟着结束。运行结果：
7、进程共享变量
共享变量不适用于多进程，进程间的变量是互相隔离的，子进程的全局变量是完全复制一份父进程的数据，对子进程的全局变量修改完全影响不到其他进程的全局变量。
示例代码：
import time
from multiprocessing import Process
def producer(a):
a += 1
time.sleep(2)
def consumer(a):
time.sleep(3)
data = a
print(data)
if __name__ == "__main__":
a = 1
my_producer = Process(target=producer, args=(a, ))
my_consumer = Process(target=consumer, args=(a, ))
my_producer.start()
my_consumer.start()
my_producer.join()
my_consumer.join()
# 输出结果为1运行结果：
示例代码： 【进程之间的变量是无法共享的，即使是全局变量也是不能共享的】
from multiprocessing import Process
import os
def func():
global n
n = 10
print('子进程pid:{},n:{}'.format(os.getppid(), n))
if __name__ == '__main__':
n = 100
print('主进程pid:{},n:{}'.format(os.getppid(), n))
p = Process(target=func)
p.start()
p.join()
print('主进程中输出n：{}'.format(n))运行结果：
8、管道Pipe(两进程间的通信优先考虑)
Pipe([duplex]):在进程之间创建一条管道，并返回元组（conn1,conn2）,其中conn1，conn2表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道。dumplex:默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，conn2只能用于发送。
- conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError。
- conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象
- conn1.close():关闭连接。如果conn1被垃圾回收，将自动调用此方法
- conn1.fileno():返回连接使用的整数文件描述符
- conn1.poll([timeout]):如果连接上的数据可用，返回True。timeout指定等待的最长时限。如果省略此参数，方法将立即返回结果。如果将timeout射成None，操作将无限期地等待数据到达。
- conn1.recv_bytes([maxlength]):接收c.send_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息，超过了这个最大值，将引发IOError异常，并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭，再也不存在任何数据，将引发EOFError异常。
- conn.send_bytes(buffer [, offset [, size]])：通过连接发送字节数据缓冲区，buffer是支持缓冲区接口的任意对象，offset是缓冲区中的字节偏移量，而size是要发送字节数。结果数据以单条消息的形式发出，然后调用c.recv_bytes()函数进行接收
- conn1.recv_bytes_into(buffer [, offset]):接收一条完整的字节消息，并把它保存在buffer对象中，该对象支持可写入的缓冲区接口（即bytearray对象或类似的对象）。offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间，将引发BufferTooShort异常。
示例代码：
import time
from multiprocessing import Process, Queue, Pool, Manager, Pipe
def producer(pipe):
pipe.send("a")
time.sleep(3)
print(pipe.recv())
def consumer(pipe):
time.sleep(2)
data = pipe.recv()
pipe.send("b")
print(data)
if __name__ == "__main__":
# Pipe实现两进程间通信
s_pipe, r_pipe = Pipe()
pool = Pool()
pool.apply_async(producer, args=(s_pipe, ))
pool.apply_async(consumer, args=(r_pipe, ))
pool.close()
pool.join()运行结果：
示例代码：
from multiprocessing import Process, Pipe
def f(conn):
conn.send('主进程，你好呀！') # 发送数据给主进程
print('子进程收到主进程发来的数据：{}'.format(conn.recv()))
conn.close() # 关闭
if __name__ == '__main__':
# Pipe是一个函数，返回的是一个元组
parent_conn, child_conn = Pipe()
# 创建一个子进程
p = Process(target=f, args=(child_conn,))
p.start()
print("主进程收到子进程发来的数据：{}".format(parent_conn.recv()))
parent_conn.send('子进程，你好啊！') # 发送数据给子进程
p.join()运行结果：
9、进程之间的同步控制
9.1 进程锁：Lock()
当多个进程对同一资源进行IO操作时，需要对资源“上锁”，否则会出现意外结果。上锁之后，同一件就只能有一个进程运行上锁的代码块。例如有一个txt文件，里面内容是一个数字10，我们要用多进程去读取这个文件的值，然后每读一次，让txt中的这个数字减1，不加锁时代码如下：
import time
import os
from multiprocessing import Process
from multiprocessing import Lock
def func():
if os.path.exists('num.txt'):
with open('num.txt', 'r') as rf:
num = int(rf.read())
num -= 1
time.sleep(1)
with open('num.txt', 'w') as wf:
wf.write(str(num))
else:
with open('num.txt', 'w') as wf:
wf.write('10')
if __name__ == '__main__':
print("主进程开始运行……")
p_list = []
for i in range(10):
p = Process(target=func)
p_list.append(p)
p.start()
for p in p_list:
p.join()
with open('num.txt', 'r') as f:
num = int(f.read())
print('最后结果为：{}'.format(num))
print("主进程结束运行……" )运行结果：
虽然用了10个进程读取并修改txt文件，但最后的值却不是1。这正是多进程共同访问资源造成混乱造成的。要达到预期结果，就要给资源上锁：
import time
import os
from multiprocessing import Process
from multiprocessing import Lock
def func(lock):
if os.path.exists('num.txt'):
lock.acquire()
with open('num.txt', 'r') as f:
num = int(f.read())
num -= 1
time.sleep(1)
with open('num.txt', 'w') as f:
f.write(str(num))
lock.release()
else:
with open('num.txt', 'w') as f:
f.write('10')
if __name__ == '__main__':
print("主进程开始运行……")
lock = Lock()
p_list = []
for i in range(10):
p = Process(target=func, args=(lock,))
p_list.append(p)
p.start()
for p in p_list:
p.join()
with open('num.txt', 'r') as f:
num = int(f.read())
print('最后结果为：{}'.format(num))
print("主进程结束运行……")运行结果：
果然，用了进程锁之后获得了预料中的结果。但是，如果你运行了上面两块代码你就会发现，加了锁之后，程序明显变慢了很多，因为程序成了串行的了，当然好处是数据安全有保证。
9.2 信号量：Semaphore
锁同时只允许一个线程更改数据，而信号量是同时允许一定数量的进程更改数据 。假如有一下应用场景：有10个人吃饭，但只有一张餐桌，只允许做3个人，没上桌的人不允许吃饭，已上桌吃完饭离座之后，下面的人才能抢占桌子继续吃饭，如果不用信号量，肯定是10人一窝蜂一起吃饭：
from multiprocessing import Process
import time
import random
def fun(i):
print('{}号顾客上座，开始吃饭'.format(i))
time.sleep(random.randint(3, 8))
print('{}号顾客吃完饭了，离座'.format(i))
if __name__ == '__main__':
for i in range(10):
p = Process(target=fun, args=(i,))
p.start()运行结果：
用了信号量，实现了轮流吃饭，每次只有3个人吃饭：
示例代码：
from multiprocessing import Process
import time
import random
from multiprocessing import Semaphore
def fun(i , sem):
sem.acquire()
print('{}号顾客上座，开始吃饭'.format(i))
time.sleep(random.randint(3, 8))
print('{}号顾客吃完饭了，离座'.format(i))
sem.release()
if __name__ == '__main__':
sem = Semaphore(3)
for i in range(10):
p = Process(target=fun, args=(i,sem))
p.start()运行结果：
事实上，Semaphore的作用也类似于锁，只不过在锁机制上添加了一个计数器，允许多个人拥有“钥匙”。
9.3 事件：Event
python进程的事件用于主进程控制其他子进程的执行，Event类有如下几个主要方法：
1）wait() 插入在进程中插入一个标记（flag）默认为 False，当 flag为False时，程序会停止运行进入阻塞状态；
2）set() 使flag为True，程序会进入非阻塞状态
3）clear() 使flag为False，程序会停止运行，进入阻塞状态
4）is_set() 判断flag 是否为True，是的话返回True，不是则返回False
有如下需求：获取当前时间的秒数的个位数，如果小于5，则设置子进程阻塞，如果大于5则设置子进程非阻塞。代码如下：
from multiprocessing import Event, Processimport timefrom datetime import datetimedef func(e): print('子进程：开始运行……') while True: print('子进程：现在事件秒数是{}'.format(datetime.now().second)) e.wait() # 阻塞等待信号 这里插入了一个flag 默认为 False time.sleep(1)if __name__ == '__main__': e = Event() p = Process(target=func, args=(e,)) p.start() for i in range(10): s = int(str(datetime.now().second)[-1]) # 获取当前秒数的个位数 if s < 5: print('子进程进入阻塞状态') e.clear() # 使插入的flag为False 进程进入阻塞状态 else: print('子进程取消阻塞状态') e.set() # 使插入的flag为True，进程进入非阻塞状态 time.sleep(1) e.set() time.sleep(3) p.terminate() print("主进程运行结束……")运行结果：
10、内置线程池
示例代码：
import time
import os
import random
from multiprocessing.pool import ThreadPool
def task():
print(f'开始执行任务：{os.getpid()}')
time.sleep(random.randint(0, 5))
print(f"执行任务结束：{os.getpid()}")
if __name__ == '__main__':
pool = ThreadPool(2)
for i in range(5):
pool.apply_async(task)
pool.close()
pool.join()运行结果：
到此这篇关于关于python多进程中的常用方法详解的文章就介绍到这了,更多相关python多进程常用方法内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
在本篇文章里小编给大家整理的是关于python设置中文界面实例方法，有兴趣的朋友们可以学习参考下。2020-10-10
这篇文章主要介绍了Python 创建格式化字符串方法，文章围绕主题展开详细的内容介绍，具有一定的参考价值，需要的小伙伴可以参考一下2022-08-08
用Python爬取LOL所有的英雄信息以及英雄皮肤的示例代码
这篇文章主要介绍了用Python爬取LOL所有的英雄信息以及英雄皮肤的示例代码，主要分为两部分，获取网页上数据和图片保存到本地等，感兴趣的可以了解一下2020-07-07
这篇文章主要给大家介绍了关于Python中对象的比较操作==和is区别的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2021-02-02
这篇文章主要介绍了Python实现抓取页面上链接的简单爬虫分享,本文使用了一个开源模块requests实现需求,需要的朋友可以参考下2015-01-01
这篇文章主要介绍了在PyCharm中使用FMEObjects,本文将ArcGISPro2.8的Python3.7解释器与FME2022.0配合使用，通过图文并茂的形式给大家介绍的非常详细，需要的朋友可以参考下2022-08-08
这篇文章主要介绍了python中查看.db文件中表格的名字及表格中的字段操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-07-07
这篇文章主要介绍了pyspark给dataframe增加新的一列的实现示例,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2020-04-04
这篇文章主要介绍了Python使用Flask库文件上传实例,用 Flask 处理文件上传很容易，只要确保HTML表单中设置enctype="multipart/form-data"属性就可以了,需要的朋友可以参考下2023-08-08
这篇文章主要介绍了Python中的 is 和 == 以及字符串驻留机制详解,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2019-06-06
最新评论