Python多进程之进程同步及通信详解
更新时间：2021年11月25日 15:48:19 作者：程序员-夏天
这篇文章主要为大家介绍了Python多进程之进程同步及通信，具有一定的参考价值，感兴趣的小伙伴们可以参考一下，希望能够给你带来帮助
上篇文章介绍了什么是进程、进程与程序的关系、进程的创建与使用、创建进程池等，接下来就来介绍一下进程同步及进程通信。
进程同步
当多个进程使用同一份数据资源的时候，因为进程的运行没有顺序，运行起来也无法控制，如果不加以干预，往往会引发数据安全或顺序混乱的问题，所以要在多个进程读写共享数据资源的时候加以适当的策略，来保证数据的一致性问题。
Lock（锁）
一个Lock对象有两个方法：acquire()和release()来控制共享数据的读写权限, 看下面这张图片，使用多进程的时候会经常出现这种情况，这是因为多个进程都在抢占输出资源，共享同一打印终端，从而造成了输出信息的错乱。
那么就可以使用Lock机制：
import multiprocessing
import random
import time
def work(lock, i):
lock.acquire()
print("work'{}'执行中......".format(i), multiprocessing.current_process().name, multiprocessing.current_process().pid)
time.sleep(random.randint(0, 2))
print("work'{}'执行完毕......".format(i))
lock.release()
if __name__ == '__main__':
lock = multiprocessing.Lock()
for i in range(5):
p = multiprocessing.Process(target=work, args=(lock, i))
p.start()
由于引入了Lock机制，同一时间只能有一个进程抢占到输出资源，其他进程等待该进程结束，锁释放到，才可以抢占，这样会解决多进程间资源竞争导致数据错乱的问题，但是由并发执行变成了串行执行，会牺牲运行效率。
进程通信
上篇文章说过，进程之间互相隔离，数据是独立的，默认情况下互不影响，那要如何实现进程间通信呢？Python提供了多种进程通信的方式，下面就来说一下。
Queue（队列）
multiprocessing模块提供的Queue多进程安全的消息队列，可以实现多进程之间的数据传递。
说明
- 初始化Queue()对象时（例如：q=Queue()），若括号中没有指定最⼤可接收的消息数量，或数量为负值，那么就代表可接受的消息数量没有上限（直到内存的尽头）。
Queue.qsize()：返回当前队列包含的消息数量。Queue.empty()：如果队列为空，返回True，反之False。Queue.full()：如果队列满了，返回True,反之False。Queue.get(block, timeout)：获取队列中的⼀条消息，然后将其从列队中移除，block默认值为True。如果block使⽤默认值，且没有设置timeout（单位秒），消息列队如果为空，此时程序将被阻塞（停在读取状态），直到从消息列队读到消息为⽌，如果设置了timeout，则会等待timeout秒，若还没读取到任何消息，则抛出Queue.Empty异常；如果block值为False，消息列队如果为空，则会⽴刻抛出Queue.Empty异常。Queue.get_nowait()：相当Queue.get(False)。Queue.put(item, block, timeout)：将item消息写⼊队列，block默认值为True，如果block使⽤默认值，且没有设置timeout（单位秒），消息列队如果已经没有空间可写⼊，此时程序将被阻塞（停在写⼊状态），直到消息列队腾出空间为⽌，如果设置了timeout，则会等待timeout秒，若还没空间，则抛出Queue.Full异常；如果block值为False，消息列队如果没有空间可写⼊，则会⽴刻抛出Queue.Full异常。Queue.put_nowait(item)：相当于Queue.put(item, False)。
from multiprocessing import Process, Queue
import time
def write_task(queue):
"""
向队列中写入数据
:param queue: 队列
:return:
"""
for i in range(5):
if queue.full():
print("队列已满!")
message = "消息{}".format(str(i))
queue.put(message)
print("消息{}写入队列".format(str(i)))
def read_task(queue):
"""
从队列读取数据
:param queue: 队列
:return:
"""
while True:
print("从队列读取:{}".format(queue.get(True)))
if __name__ == '__main__':
print("主进程执行......")
# 主进程创建Queue，最大消息数量为3
queue = Queue(3)
pw = Process(target=write_task, args=(queue, ))
pr = Process(target=read_task, args=(queue, ))
pw.start()
pr.start()
运行结果为：
从结果我们可以看出，队列最大可以放入3条消息，后面再来消息，要等read_task从队列里取出后才行。
Pipe（管道）
Pipe常用于两个进程，两个进程分别位于管道的两端，Pipe(duplex)方法返回（conn1,conn2）代表一个管道的两端，duplex参数默认为True，即全双工模式，若为False，conn1只负责接收信息，conn2负责发送。
send()和recv()方法分别是发送和接受消息的方法。
import multiprocessing
import time
import random
def proc_send(pipe):
"""
发送消息
:param pipe:管道一端
:return:
"""
for i in range(10):
print("process send:{}".format(str(i)))
pipe.send(i)
time.sleep(random.random())
def proc_recv(pipe):
"""
接收消息
:param pipe:管道一端
:return:
"""
while True:
print("Process recv:{}".format(pipe.recv()))
time.sleep(random.random())
if __name__ == '__main__':
# 主进程创建pipe
pipe = multiprocessing.Pipe()
p1 = multiprocessing.Process(target=proc_send,args=(pipe[0], ))
p2 = multiprocessing.Process(target=proc_recv,args=(pipe[1], ))
p1.start()
p2.start()
p1.join()
p2.terminate()
执行结果为：
Semaphore（信号量）
Semaphore用来控制对共享资源的访问数量，和进程池的最大连接数类似。
import multiprocessing
import random
import time
def work(s, i):
s.acquire()
print("work'{}'执行中......".format(i), multiprocessing.current_process().name, multiprocessing.current_process().pid)
time.sleep(i*2)
print("work'{}'执行完毕......".format(i))
s.release()
if __name__ == '__main__':
s = multiprocessing.Semaphore(2)
for i in range(1, 7):
p = multiprocessing.Process(target=work, args=(s, i))
p.start()
上面的代码中使用Semaphore限制了最多有2个进程同时执行，那么来一个进程获得一把锁，计数加1，当计数等于2时，后面再来的进程均需要等待，等前面的进程释放掉，才可以获得锁。
信号量与进程池的概念上类似，但是要区分开来，信号量涉及到加锁的概念。
Event（事件）
Event用来实现进程间同步通信的。运行的机制是：全局定义了一个flag，如果flag值为False，当程序执行event.wait()方法时就会阻塞，如果flag值为True时，程序执行event.wait()方法时不会阻塞继续执行。
Event常⽤函数：
event.wait()：在进程中插入一个标记（flag），默认为False，可以设置timeout。event.set()：使flag为Ture。event.clear()：使flag为False。event.is_set()：判断flag是否为True。
import multiprocessing
import time
def wait_for_event(e):
print("wait_for_event执行")
e.wait()
print("wait_for_event: e.is_set():{}".format(e.is_set()))
def wait_for_event_timeout(e, t):
print("wait_for_event_timeout执行")
# 只会阻塞2s
e.wait(t)
print("wait_for_event_timeout:e.is_set:{}".format(e.is_set()))
if __name__ == "__main__":
e = multiprocessing.Event()
p1 = multiprocessing.Process(target=wait_for_event, args=(e,))
p1.start()
p2 = multiprocessing.Process(target=wait_for_event_timeout, args=(e, 2))
p2.start()
time.sleep(4)
# 4s之后使用e.set()将flag设为Ture
e.set()
print("主进程:flag设置为True")
执行结果如下：
总结
本篇文章就到这里了，希望能够给你带来帮助，也希望您能够多多关注脚本之家的更多内容!
相关文章
这篇文章主要介绍了python无法识别vim中文代码的解决方案，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2021-03-03
python利用蒙版抠图(使用PIL.Image和cv2)输出透明背景图
这篇文章主要介绍了python利用蒙版抠图(使用PIL.Image和cv2)输出透明背景图,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2020-08-08
这篇文章主要介绍了Python爬虫网页元素定位术，文章通过Beautiful Soup模块展开详细的内容介绍，具有一定的参考价值，需要的小伙伴可以参考一下2022-05-05
这篇文章主要介绍了Python的三种可变参数，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2019-05-05
这篇文章主要介绍了Python基于sklearn库的分类算法,结合简单实例形式分析了Python使用sklearn库封装朴素贝叶斯、K近邻、逻辑回归、SVM向量机等常见机器学习算法的分类调用相关操作技巧,需要的朋友可以参考下2018-07-07
这篇文章主要介绍了利用Python实现批量解压&压缩文件夹的示例代码，文中的实现步骤讲解详细，感兴趣的小伙伴快跟随小编一起动手试一试2022-04-04
在本篇文章里小编给大家整理了关于学习和使用python的13个理由以及相关知识点，需要的朋友们参考下。2019-07-07
今天小编就为大家分享一篇python科学计算之numpy——ufunc函数用法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2019-11-11
本问重点讲解了Python文件的读写原理、常用文件打开模式、文件对象的常用方法和目录的相关操作详细讲解，对Python文件操作相关知识感兴趣的朋友跟随小编一起看看吧2021-04-04
2022年奥林匹克运动会如期举行，以不正确的方式进行运动风险在增加，人体姿势估计是计算机视觉领域的重要问题，接下来通过本文给大家介绍下使用OpenCV对运动员的姿势进行检测功能，感兴趣的朋友一起看看吧2022-02-02
最新评论