此前的几篇文章中,我们介绍了 python 进程间通信的一系列方案: python 进程间通信(一) — 信号的基本使用 python 进程间通信(二) — 定时信号 SIGALRM python 进程间通信(三) — 进程同步原语及管道与队列
回顾操作系统所提供的所有进程间通信方式的系统调用,我们会发现还有两种进程间通信方式我们还没有介绍:共享内存与域套接字,本文我们就来介绍这剩下的几种 IPC 方式。
通常,在并发环境下应该尽量避免数据和状态的共享,因为这意味着竞争条件的产生,而进程间的同步就意味着效率的降低以及更高的复杂度。 但 Python 的 multiprocessing 包中仍然提供了两种方法让你可以在多进程环境下共享数据:
共享内存是进程间共享数据最简单的方式,python 中有两个方法来创建共享的数据对象,分别是:
对于 Value 对象,我们需要通过他的 value 字段获取到实际的值,而 Array 对象则可以直接通过下标访问元素。
typecode_or_type 既可以是一个描述类型的字符串,也可以是一个ctypes 包中定义的枚举。 下表列出了可以选取的取值:
ctypes 枚举 | 字符串 | 说明 |
---|---|---|
py_object | ‘O’ | python 对象 |
c_short | ‘h’ | 系统中的 short 类型 |
c_ushort | ‘H’ | 系统中的 ushort 类型 |
c_long | ‘l’ | 系统中的 long 类型 |
c_ulong | ‘L’ | 系统中的 ulong 类型 |
c_int | ‘i’ | 系统中的 int 类型 |
c_uint | ‘I’ | 系统中的 uint 类型 |
c_float | ‘f’ | 系统中的 float 类型 |
c_double | ‘d’ | 系统中的 double 类型 |
c_longdouble | ‘g’ | 系统中的 longdouble 类型 |
c_longlong | ‘q’ | 系统中的 longlong 类型 |
c_ulonglong | ‘Q’ | 系统中的 ulonglong 类型 |
c_byte | ‘b’ | 系统中的 byte 类型 |
c_ubyte | ‘B’ | 系统中的 ubyte 类型 |
c_char | ‘c’ | 系统中的 char 类型 |
c_char_p | ‘z’ | 系统中的NUL结尾字符串 |
c_wchar_p | ’Z’ | 系统中的 unicode NUL 结尾字符串 |
c_bool | ‘?’ | 系统中的 bool 类型 |
使用共享数据,就必然涉及到竞争条件的抢夺,普通的赋值、加减乘除都是原子性的,但有时我们需要执行一些并不是原子性的操作,此时就需要加锁,例如先比较后操作,特别的,一个最容易忽略的例子是 += 操作,很容易被认为是一个原子操作,事实上,他是加操作与赋值操作的结合,并不是一个原子操作。 对一个共享内存进行非原子的一系列操作就要考虑加锁,通过将锁对象传递给 lock 参数,我们可以通过共享内存对象的 get_lock 方法获取并使用该锁对象。 lock 参数的默认值是 True,python 解释器会选取系统所支持的锁来创建一个锁对象,如果传递 False,则表示不创建锁。
from ctypes import c_double
from multiprocessing import Process, Value, Array
def f(n, a):
n.value = 3.1415927
for i in range(len(a)):
a[i] = -a[i]
if __name__ == '__main__':
num = Value(c_double, 0.0)
arr = Array('i', range(10))
p1 = Process(target=f, args=(num, arr))
p1.start()
p2 = Process(target=f, args=(num, arr))
p2.start()
p1.join()
p2.join()
print(num.value)
print(arr[:])
上面的例子中,在主进程与子进程间共享了一个 double 类型的数字和一个 int 型数组,最终打印出被子进程修改的最终值:
3.1415927 [0, -1, -2, -3, -4, -5, -6, -7, -8, -9]
from multiprocessing import Process, Value
def func(n):
if n.value <= 10:
n.value += 1
if __name__ == '__main__':
num = Value('i', 0)
processes = []
for _ in range(50):
processes.append(Process(target=func, args=[num]))
for process in processes:
process.start()
for process in processes:
process.join()
print(num.value)
打印出了:
13
上述代码非常简单,创建了 10 个进程并发处理,每个进程中先判断共享内存中数字的值,如果该值不大于 10 则进行加 1 操作。 理论上, 数字是不会被加到 11 以上的,但是实际打印出的数字却是 12,且多次执行结果会出现不同,这是为什么呢? 假设共享内存中数字为 10,多个进程同时判断该共享内存中的数字是否不大于 10 均返回 True,于是他们都对共享内存中的数字进行加 1 操作,就出现了实际执行 +1 的次数超过了预期次数。
解决这样的问题的方法就是加锁:
from multiprocessing import Process, Value
def func(n):
with n.get_lock():
if n.value <= 10:
n.value += 1
if __name__ == '__main__':
num = Value('i', 0)
processes = []
for _ in range(50):
process = Process(target=func, args=[num])
processes.append(process)
process.start()
for process in processes:
process.join()
print(num.value)
稳定打印出了:
11
python 提供了一种十分类似共享内存的数据共享机制 — 服务器进程。 通过 multiprocessing 包中的 Manager 类可以构造一个服务器进程对象,他支持用于进程间共享的多种数据类型:
一旦创建,对象的使用与原生类型的用法是完全相同的,因此相比于共享内存,服务器进程的使用更为简单和灵活,但由于实现更为复杂,运行效率略低于共享内存。
from multiprocessing import Process, Manager
def f(d, l):
d[1] = '1'
d['2'] = 2
d[0.25] = None
l.reverse()
if __name__ == '__main__':
with Manager() as manager:
d = manager.dict()
l = manager.list(range(10))
p = Process(target=f, args=(d, l))
p.start()
p.join()
print(d)
print(l)
打印出了。
{0.25: None, 1: ’1’, ’2’: 2} [9, 8, 7, 6, 5, 4, 3, 2, 1, 0]