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

1. 引言

此前的几篇文章中，我们介绍了 python 进程间通信的一系列方案： python 进程间通信（一） — 信号的基本使用 python 进程间通信（二） — 定时信号 SIGALRM python 进程间通信（三） — 进程同步原语及管道与队列

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

2. 并发环境下的数据共享

通常，在并发环境下应该尽量避免数据和状态的共享，因为这意味着竞争条件的产生，而进程间的同步就意味着效率的降低以及更高的复杂度。 但 Python 的 multiprocessing 包中仍然提供了两种方法让你可以在多进程环境下共享数据：

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

3. 共享内存

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

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

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

3.1. typecode_or_type 参数

typecode_or_type 既可以是一个描述类型的字符串，也可以是一个ctypes 包中定义的枚举。 下表列出了可以选取的取值：

typecode_or_type 参数取值

3.2. lock 参数

使用共享数据，就必然涉及到竞争条件的抢夺，普通的赋值、加减乘除都是原子性的，但有时我们需要执行一些并不是原子性的操作，此时就需要加锁，例如先比较后操作，特别的，一个最容易忽略的例子是 += 操作，很容易被认为是一个原子操作，事实上，他是加操作与赋值操作的结合，并不是一个原子操作。 对一个共享内存进行非原子的一系列操作就要考虑加锁，通过将锁对象传递给 lock 参数，我们可以通过共享内存对象的 get_lock 方法获取并使用该锁对象。 lock 参数的默认值是 True，python 解释器会选取系统所支持的锁来创建一个锁对象，如果传递 False，则表示不创建锁。

3.3. 示例

3.3.1. 进程间通过共享内存共享数据

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]

3.3.2. 使用锁对象保证共享数据的安全性

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

4. 服务器进程 — server process

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

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

4.1. 示例

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]