qidao123.com技术社区-IT企服评测·应用市场

标题: 一文速通Python并行计算：09 Python多进程编程-进程之间的数据同步-基于互斥锁、递归锁、信号量、条件变量、事故和屏蔽 [打印本页]

作者: 半亩花草 时间: 2025-4-22 23:44
标题: 一文速通Python并行计算：09 Python多进程编程-进程之间的数据同步-基于互斥锁、递归锁、信号量、条件变量、事故和屏蔽
一文速通 Python 并行计算：09 Python 多进程编程-进程之间的数据同步-基于互斥锁、递归锁、信号量、条件变量、事故和屏蔽

摘要：

多进程同步机制包括互斥锁、递归锁、信号量、条件变量、事故和屏蔽等：互斥锁用于掩护共享资源，递归锁支持同一进程重复加锁，信号量可控制访问资源的数量，条件变量用于进程间等待特定条件，事故用于进程间通信和同步，屏蔽用于多个进程在特定点同步，确保协同运行。

关于我们更多介绍可以查看云文档：Freak 嵌入式工作室云文档，大概访问我们的 wiki：****https://github.com/leezisheng/Doc/wik

正文

进程之间数据除了共享内存、共享进程外不共享,但是共享同一套文件体系,所以访问同一个文件,或同一个打印终端,是没有问题的,而共享带来的是竞争，竞争带来的结果就是错乱，如何控制，需要进程之间进行同步。
进程的同步原语包括：Lock、Event、Condition、Semaphore、Rlock、Barrier。相关的同步原语和线程的库很雷同。
如下代码是多个进程差别步共享同一打印终端，采取并发运行,服从高,但竞争同一打印终端,带来了打印错乱。

from multiprocessing import Process
import os,time
def work():
print('%s is running' %os.getpid())
time.sleep(2)
print('%s is done' %os.getpid())
if __name__ == '__main__':
for i in range(3):
p=Process(target=work)
p.start()

复制代码

以下为运行结果，可以看到并应该是三个进程轮流输出的两条语句变为乱序输出。

1.基于互斥锁的进程数据同步

如下 work 函数中利用 acquire() 和 release() ，来控制共享数据的读写权限：

from multiprocessing import Process,Lock
import os,time
def work(lock):
lock.acquire()
print('%s is running' %os.getpid())
time.sleep(2)
print('%s is done' %os.getpid())
lock.release()
if __name__ == '__main__':
lock=Lock()
for i in range(3):
p=Process(target=work,args=(lock,))
p.start()

复制代码

可以看到按序输出，加锁实现由并发变成了串行,牺牲了运行服从,但避免了竞争。

2.基于递归锁的进程数据同步

示例代码如下：

from multiprocessing import Process,RLock
import multiprocessing
import time
mutex = RLock()
def test(mutex):
if mutex.acquire():
print("I am %s" % multiprocessing.current_process().name)
if mutex.acquire():
print("I am %s" % multiprocessing.current_process().name)
mutex.release()
time.sleep(1)
mutex.release()
if __name__ == '__main__':
for i in range(0, 10):
p = Process(target=test,args=(mutex,))
p.start()

复制代码

输出如下，可以看到输出正常：

3.基于信号量的进程数据同步

以下为示例代码，从运行结果中我们可以看到两个 print 依次输出。

from multiprocessing import Process,Semaphore
import multiprocessing
import time
_# 创建一个信号量，信号量是一个内部数据_
_# 用于标明当前的共享资源可以有多少并发读取_
semaphore = Semaphore(1)
def test(semaphore):
_# 测试控制该资源的信号量。_
if semaphore.acquire():
_# 若此信号量的值为正，则允许进行使用该资源。线程将信号量减 1。_
print("I am %s,pid:%s" % (multiprocessing.current_process().name,multiprocessing.current_process().pid))
print("%s am runing" % multiprocessing.current_process().name)
semaphore.release()
if __name__ == '__main__':
for i in range(0, 10):
p = Process(target=test,args=(semaphore,))
p.start()

复制代码

4.基于条件变量的进程数据同步

这里，我们以生产者消费者模型为例，在多线程的基于条件变量的线程同步示例中，我们利用全局变量看成缓存，在本例中，我们同样如此。
有些同学会问，不是说进程中的数据是独立的吗？没错，在多进程编程中，差别的进程之间默认情况下是无法共享数据的。但 Python 提供了一些机制来实现多进程间的数据共享，其中之一是共享内存。共享内存允许多个进程共享一个存储地区，一个进程写入共享内存中的信息，其他进程可以方便的读取。在 Python 中可以利用 Value、Array 将数据存储在共享内存中，也可以利用 multiprocessing 模块中 sharedctypes 自定义共享内存的 ctypes 对象。

下例中，以 value 对象作为缓存，只要缓存不满，生产者一直向缓存生产；只要缓存不空，消费者一直从缓存取出（之后烧毁）。
当缓冲队列不为空的时候，生产者将通知消费者；当缓冲队列不满的时候，消费者将通知生产者。

from multiprocessing import Process,Condition,Value
import multiprocessing
import time
condition = Condition()
products = Value('i', 0)
_# 生产者进程_
def Producer(condition,products):
while True:
if condition.acquire():
_# 消费者通过拿到锁来修改共享的资源_
if products.value < 10:
_# 如果产品数量小于 10，继续生成，并通过 notify 方法通知消费者_
_# 只要缓存不满，生产者一直向缓存生产；_
products.value += 1;
print("Producer(%s):deliver one, now products:%s" % (multiprocessing.current_process().name, products.value))
_# 当缓冲队列不为空的时候，生产者将通知消费者_
condition.notify()
else:
_# 如果已经满了，那么生产者进入等待状态，直到被唤醒_
print("Producer(%s):already 10, stop deliver, now products:%s" %(multiprocessing.current_process().name, products.value))
condition.wait()
_# 释放资源_
condition.release()
time.sleep(1)
_# 消费者进程_
def Consumer(condition,products):
while True:
if condition.acquire():
if products.value > 1:
_# 只要缓存不空，消费者一直从缓存取出（之后销毁）_
products.value -= 1
print("Consumer(%s):consume one, now products:%s" % (multiprocessing.current_process().name, products.value))
_# 当缓冲队列不满的时候，消费者将通知生产者。_
condition.notify()
else:
_# 缓存空，消费者线程等待_
print("Consumer(%s):only 1, stop consume, products:%s" % (multiprocessing.current_process().name, products.value))
condition.wait()
_# 释放资源_
condition.release()
time.sleep(1)
if __name__ == '__main__':
ProducerProcess = Process(target=Producer,args=(condition,products))
ConsumerProcess = Process(target=Consumer, args=(condition, products))
ProducerProcess.start()
ConsumerProcess.start()

复制代码

如下为运行情况，可以看到正常运行。

5.基于事故的进程数据同步

示例代码和运行结果如下，与多线程中雷同。

from multiprocessing import Process,Event
import multiprocessing
event = Event()
def worker(event_obj, i):
print('{i}号进程等待事件信号'.format(i=i))
event_obj.wait()
print('{i}号进程收到事件信号'.format(i=i))
if __name__ == '__main__':
for i in range(5):
p = Process(target=worker, args=(event, i))
p.start()
event.set()

复制代码

6.基于屏蔽的进程数据同步

下面的代码展示了如何利用 barrier() 函数来同步两个进程。我们有 4 个进程，进程 1 和进程 2 由 barrier 语句管理，进程 3 和进程 4 没有同步策略。

import multiprocessing
from multiprocessing import Barrier, Lock, Process
import time
from datetime import datetime
def test_with_barrier(synchronizer, serializer):
name = multiprocessing.current_process().name
synchronizer.wait()
now = time.time()
time.sleep(1)
with serializer:
print("process %s ----> %s" % (name, datetime.fromtimestamp(now)))
def test_without_barrier():
name = multiprocessing.current_process().name
now = time.time()
print("process %s ----> %s" % (name, datetime.fromtimestamp(now)))
if __name__ == '__main__':
synchronizer = Barrier(2)
serializer = Lock()
Process(name='p1 - test_with_barrier', target=test_with_barrier, args=(synchronizer,serializer)).start()
Process(name='p2 - test_with_barrier', target=test_with_barrier, args=(synchronizer,serializer)).start()
Process(name='p3 - test_without_barrier', target=test_without_barrier).start()
Process(name='p4 - test_without_barrier', target=test_without_barrier).start()

复制代码

运行代码，将看到进程 1 和进程 2 在同一时间打印：

下面这幅图表示了 barrier 如何同步两个进程：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/)