python 进程间通信(四) -- 共享内存与服务器进程

2019-06-20 00:02:11   最后更新: 2019-06-20 00:02:11   访问数量:43




此前的几篇文章中,我们介绍了 python 进程间通信的一系列方案:

python 进程间通信(一) -- 信号的基本使用

python 进程间通信(二) -- 定时信号 SIGALRM

python 进程间通信(三) -- 进程同步原语及管道与队列

 

回顾操作系统所提供的所有进程间通信方式的系统调用,我们会发现还有两种进程间通信方式我们还没有介绍:共享内存与域套接字,本文我们就来介绍这剩下的几种 IPC 方式

 

 

通常,在并发环境下应该尽量避免数据和状态的共享,因为这意味着竞争条件的产生,而进程间的同步就意味着效率的降低以及更高的复杂度

但 Python 的 multiprocessing 包中仍然提供了两种方法让你可以在多进程环境下共享数据:

  1. 共享内存
  2. 服务器进程

 

共享内存是进程间共享数据最简单的方式,python 中有两个方法来创建共享的数据对象,分别是:

  1. Value(typecode_or_type, *args, lock=True) -- 开辟共享内存空间存储值类型
  2. Array(typecode_or_type, size_or_initializer, *, lock=True) -- 开辟共享内存空间存储数组类型

 

对于 Value 对象,我们需要通过他的 value 字段获取到实际的值,而   Array 对象则可以直接通过下标访问元素

 

typecode_or_type 参数

typecode_or_type 既可以是一个描述类型的字符串,也可以是一个ctypes 包中定义的枚举

下表列出了可以选取的取值:

typecode_or_type 参数取值
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 参数

使用共享数据,就必然涉及到竞争条件的抢夺,普通的赋值、加减乘除都是原子性的,但有时我们需要执行一些并不是原子性的操作,此时就需要加锁,例如先比较后操作,特别的,一个最容易忽略的例子是 += 操作,很容易被认为是一个原子操作,事实上,他是加操作与赋值操作的结合,并不是一个原子操作

对一个共享内存进行非原子的一系列操作就要考虑加锁,通过将锁对象传递给 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 类可以构造一个服务器进程对象,他支持用于进程间共享的多种数据类型:

  1. list
  2. dict
  3. Namespace
  4. Lock
  5. RLock
  6. Semaphore
  7. BoundedSemaphore
  8. Condition
  9. Event
  10. Barrier
  11. Queue
  12. Value
  13. Array

 

一旦创建,对象的使用与原生类型的用法是完全相同的,因此相比于共享内存,服务器进程的使用更为简单和灵活,但由于实现更为复杂,运行效率略低于共享内存

 

示例

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]

 

欢迎关注微信公众号,以技术为主,涉及历史、人文等多领域的学习与感悟,每周三到七篇推文,全部原创,只有干货没有鸡汤

 

 






读书笔记      技术帖      python      进程间通信      ipc     


京ICP备15018585号