python中的进程间通信

进程间数据是否共享

在Python中，进程之间默认是不共享内存的。每个进程都有自己独立的内存空间，这意味着在一个进程中对数据的修改不会影响到另一个进程中的同名数据。然而，Python提供了几种方式来实现进程间的数据共享：

1. 使用 multiprocessing 模块: Python 的 multiprocessing 模块提供了多种方式来实现进程间通信（IPC），其中包括：

   • Value 和 Array: 可以用于存储数据并可以被多个进程共享。
   • Manager: 通过使用 manager 对象，可以创建一个运行在服务器进程中的管理器对象，该对象支持将列表、字典等放在多个进程之间共享。

2. 使用文件或数据库: 进程可以通过读写文件或操作数据库来交换信息。这种方法相对简单但可能受到I/O性能限制。

3. 使用消息传递机制：如队列（Queue）和管道（Pipe），这些也是 multiprocessing 提供的功能。它们允许将消息从一个进程传递到另一个，虽然严格意义上不是共享内存，但它们是进行数据交换和任务协调非常有效的手段。

下面是一个简单示例代码说明，在不使用Value的情况下

from multiprocessing import Process, Lock, current_process
import time


def process_with_shared_resource(shared_resource, lock):
    name = current_process().name
    lock.acquire()
    try:
        print(f"{name} 获取当前共享变量： {shared_resource}")
        shared_resource += 1
        time.sleep(0.1)
        print(f"{name} 更新后的共享变量： {shared_resource}")
    finally:
        lock.release()


if __name__ == "__main__":
    # 创建一个共享的资源和锁
    shared_resource = 0
    lock = Lock()

    # 创建多个进程
    processes = []
    for _ in range(5):
        p = Process(target=process_with_shared_resource, args=(shared_resource, lock))
        processes.append(p)
        p.start()

    # 等待所有进程完成
    for p in processes:
        p.join()

    print(f"最终变量值: {shared_resource}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

输出：

Process-3 获取当前共享变量： 0
Process-3 更新后的共享变量： 1
Process-1 获取当前共享变量： 0
Process-1 更新后的共享变量： 1
Process-2 获取当前共享变量： 0
Process-2 更新后的共享变量： 1
Process-4 获取当前共享变量： 0
Process-4 更新后的共享变量： 1
Process-5 获取当前共享变量： 0
Process-5 更新后的共享变量： 1
最终变量值: 0
1
2
3
4
5
6
7
8
9
10
11

看到，每个进程都有自己的变量shared_resource，并且初始化值都为0。

优化使用Value后：

from multiprocessing import Process, Lock, current_process, Value
import time


def process_with_shared_resource(shared_resource, lock):
    name = current_process().name
    lock.acquire()
    try:
        print(f"{name} 获取当前共享变量： {shared_resource.value}")
        shared_resource.value += 1
        time.sleep(0.1)
        print(f"{name} 更新后的共享变量： {shared_resource.value}")
    finally:
        lock.release()


if __name__ == "__main__":
    # 创建一个共享的资源和锁
    shared_resource = Value('i', 0)
    lock = Lock()

    # 创建多个进程
    processes = []
    for _ in range(5):
        p = Process(target=process_with_shared_resource, args=(shared_resource, lock))
        processes.append(p)
        p.start()

    # 等待所有进程完成
    for p in processes:
        p.join()

    print(f"最终变量值: {shared_resource.value}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

输出：

Process-1 获取当前共享变量： 0
Process-1 更新后的共享变量： 1
Process-4 获取当前共享变量： 1
Process-4 更新后的共享变量： 2
Process-5 获取当前共享变量： 2
Process-5 更新后的共享变量： 3
Process-2 获取当前共享变量： 3
Process-2 更新后的共享变量： 4
Process-3 获取当前共享变量： 4
Process-3 更新后的共享变量： 5
最终变量值: 5
1
2
3
4
5
6
7
8
9
10
11

结果正确了

进程间通信

在前面的学习中我们提到了一个概念，就是主进程的代码执行结束以后，主进程并没有结束的，因为主进程需要等待所有子进程运行代码结束以后，主进程通过系统调用回收子进程的资源，紧接着主进程才进行系统调用回收当前进程的资源。

那么，主进程怎么在数据隔离的情况下知道每一个子进程是什么时候结束的呢？

注意：子进程是完全有可能存在input，recv这样的阻塞代码情况的。实际上，父进程不可能预判到每个子进程什么时候结束的，但是可以让子进程在结束的时候发出一个信号告诉父进程，它结束了。父进程接受到该子进程的结束信号就可以通过系统调用回收子进程的资源了。而这个发出信号与接收信号的过程，就是进程间的通信（IPC）了。

进程间通信（Inter-Process Communication，IPC）是操作系统中允许不同进程之间交换信息的一种机制。在多任务操作系统中，IPC对于实现不同程序组件之间的协作至关重要。由于每个进程都有自己独立的内存空间，所以他们不能直接共享内存中的数据。Python提供了多种机制来实现进程间的通信，主要通过 multiprocessing 模块来完成。

1. 管道（Pipes）

管道是一种最基本的进程间通信方式。在Python中，multiprocessing 模块的 Pipe() 函数可以创建一对连接对象，这两个对象可以在不同进程间通过发送和接收消息来通信。管道可以是单向的（半双工）或双向的（全双工）。

1. 创建管道：调用multiprocessing.Pipe()将创建一对可以用于进程间通信的文件描述符。

2. 读写通信：一个进程使用一个文件描述符的一端来发送数据，另一个进程使用另一个文件描述符的另一端来接收数据。

3. 半双工通信：数据只能在一个方向上流动，这意味着发送方和接收方不能同时发送数据，但可以交替进行。

4. 关闭管道：通信完成后，应该关闭管道以释放系统资源。

from multiprocessing import Process, Pipe, current_process


def child(conn):
    str_send = "Hello ~"
    print(f'{current_process().name} 发送消息：{str_send}')
    conn.send(str_send)
    conn.send(str_send)
    conn.close()


if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=child, args=(child_conn,))
    p.start()

    # recv会阻塞调用，直到收到消息
    print(f"{current_process().name} 接受到消息：{parent_conn.recv()}")
    print(f"{current_process().name} 接受到消息：{parent_conn.recv()}")
    # print(f"{current_process().name} 接受到消息：{parent_conn.recv()}") # 阻塞接受消息

    parent_conn.close()  # 关闭父进程端的管道
    p.join()  # 等待子进程结束

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

• 我们定义了一个child函数，它将通过管道发送一条消息。
• 在主进程中，我们创建了一个管道，并将管道的两个端点分别赋值给parent_conn和child_conn。
• 我们创建了一个子进程，并将child_conn传递给它。
• 子进程开始运行，并调用child函数，通过管道发送了一条消息。
• 主进程使用parent_conn.recv()来接收消息，然后打印这条消息。
• 最后，我们关闭了管道，并等待子进程结束。

注意，multiprocessing.Pipe()主要用于进程间的通信，而不是线程间的通信。

2. 队列（Queues）

队列是用于多个生产者（发送者）和消费者（接收者）的典型场景。它是线程和进程安全的。通过使用 multiprocessing.Queue，可以在进程间安全地交换消息或其他数据。

from multiprocessing import Process, Queue, current_process


def worker(queue):
    while True:
        args = queue.get()  # 阻塞调用，直到队列中有数据
        # 这里可以处理接收到的参数
        print(f"{current_process().name} 接受到消息到： {args}")
        if args is None:  # 接收到退出信号
            break


if __name__ == '__main__':
    # 创建一个队列
    q = Queue()

    # 启动一个工作进程
    p = Process(target=worker, args=(q,))
    p.start()

    # 将数据放入队列
    for i in range(10):
        print(f"{current_process().name} 发送消息： {i}")
        q.put(i)

    q.put(None)

    # 等待队列中的所有任务被处理完成
    p.join()

    print("结束")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

3. 共享内存

multiprocessing 模块还支持通过共享内存来进行进程间通信，主要通过 Value 或 Array。这允许创建一个可以在多个进程间共享的变量。

最开始的代码例子便使用的是Value模块

4. 管理器（Managers）

multiprocessing 提供了一个 Manager() 方法，用于创建一个管理器对象，该对象支持将Python对象放在服务器进程中，以便不同进程可以通过代理的方式访问。

from multiprocessing import Process, Manager, current_process


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))
        print(f"{current_process().name} dict： {d}")
        print(f"{current_process().name} list： {l}")

        p = Process(target=f, args=(d, l))
        p.start()
        p.join()

        print(f"{current_process().name} dict： {d}")
        print(f"{current_process().name} list： {l}")
        # 子进程已经修改了变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

输出：

MainProcess dict： {}
MainProcess list： [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
MainProcess dict： {1: '1', '2': 2, 0.25: None}
MainProcess list： [9, 8, 7, 6, 5, 4, 3, 2, 1, 0]
1
2
3
4

进程间的数据隔离

进程间的数据隔离是操作系统设计中的一个核心概念，它确保了不同进程的内存空间和数据是相互独立的，从而防止了进程间的非授权数据访问和潜在的数据冲突。以下是实现进程间数据隔离的一些关键机制：

1. 独立的地址空间：

   • 每个进程都有自己独立的虚拟地址空间，这意味着一个进程的代码和数据不能直接被另一个进程访问。

2. 内存保护：

   • 操作系统通过内存保护机制防止进程访问不属于它的内存区域。如果一个进程尝试访问另一个进程的内存，操作系统将触发一个保护错误（通常是段错误）。

3. 分页机制：

   • 操作系统使用分页机制来管理内存，每个进程的页表是独立的，映射到物理内存的地址也是不同的。

4. 执行上下文：

   • 每个进程都有自己的执行上下文，包括程序计数器、堆栈指针、寄存器集合等，这些上下文在进程切换时会保存和恢复。

5. 文件描述符和文件权限：

   • 进程对文件的访问是通过文件描述符进行的，操作系统通过文件权限来控制进程对文件的访问。

6. 系统调用：

   • 进程需要通过系统调用来请求操作系统提供的服务，如内存分配、文件操作等，这些系统调用会进行必要的安全检查。

7. 用户和组ID：

   • 每个进程都有一个用户ID（UID）和一个或多个组ID（GID），操作系统使用这些ID来控制进程对资源的访问权限。

8. 命名空间：

   • 现代操作系统（如Linux）使用命名空间来提供更细粒度的隔离，包括PID命名空间、网络命名空间、挂载命名空间等。

9. 安全模块和沙箱：

   • 操作系统和应用程序可以使用安全模块和沙箱技术来进一步限制进程的权限和行为。

10. 权限控制：

    • 进程的权限受到其用户权限和组权限的限制，操作系统会根据这些权限来控制进程可以执行的操作。

11. AppArmor或SELinux：

    • 这些是Linux系统中的强制访问控制系统，它们可以提供更细粒度的访问控制和隔离。