进程间通信详解

Python
495
0
0
2022-12-08
6 个最好的 Python IDE 和代码编辑器
上一篇
Python中的GIL机制详解
下一篇

进程间通信（IPC，InterProcess Communication）是指在不同进程之间传播或交换信息。 IPC的方式通常有管道（包括无名管道和命名管道）、消息队列、信号量、共享内存、Socket、Streams等。其中 Socket和Streams支持不同主机上的两个进程IPC。

一、管道

管道，通常指无名管道，是 UNIX 系统IPC最古老的形式。

1、特征

  • 它是半双工的（即数据只能在一个方向上流动），具有固定的读端和写端。
  • 它只能用于具有亲缘关系的进程之间的通信（也是父子进程或者兄弟进程之间）。
  • 它可以看成是一种特殊的文件，对于它的读写也可以使用普通的read、write等函数。但是它不是普通的文件，并不属于其他任何文件系统，并且只存在于内存中。

2、原型

#include <unistd.h>
int pipe(int fd[2]);    // 返回值：若成功返回0，失败返回-1

当一个管道建立时，它会创建两个文件描述符：fd[0]为读而打开，fd[1]为写而打开。

3、示例

这里使用python示例如下:

from multiprocessing import Pipe
from multiprocessing import Process


def sub_process(out_pipe, in_pipe):
    # 如果不关闭输入子进程的输入端，子进程会一直等待
    in_pipe.close()
    while True:
        try:
            # 子进程从管道中获取消息
            msg = out_pipe.recv()
        except EOFError:
            break 
        else:
            print(msg)


if __name__ == '__main__':
    out_pipe, in_pipe = Pipe(True)
    # 将管道的输入端和输出端传入子进程，这样父子进程共享一个管道，一端连着主进程，一端连着子进程
    sub_proc = Process(target=sub_process, args=(out_pipe, in_pipe))
    sub_proc.start()

    # 关闭主进程的输出端
    out_pipe.close()
    for i in range(5):
        # 主进程向管道中发送消息
        in_pipe.send(i)

    # 如果不关闭主进程的输入端，则子进程会一直等待
    in_pipe.close()

    # 等待子进程结束
    sub_proc.join()
    print("master proc end")

二、FIFO

FIFO，也称为命名管道，它是一种文件类型。

1、特征

  • FIFO可以在无关的进程之间交换数据，与无名管道不同。
  • FIFO有路径名与之相关联，它以一种特殊设备文件形式存在于文件系统中。

2、原型

#include <sys/stat.h>
// 返回值：成功返回0，出错返回-1
int mkfifo(const char *pathname, mode_t mode);

其中的 mode 参数与open函数中的 mode 相同。一旦创建了一个 FIFO，就可以用一般的文件I/O函数操作它。

FIFO的通信方式类似于在进程中使用文件来传输数据，只不过FIFO类型文件同时具有管道的特性。在数据读出时，FIFO管道中同时清除数据，并且“先进先出”。

3、示例

示例代码如下：

import os
import errno

pipe_file_path = "/tmp/fifo_pipe"
if __name__ == '__main__':
    try:
        os.mkfifo(pipe_file_path)
    except OSError as oe:
        if oe.errno != errno.EEXIST:
            raise

    print("open fifo pipe")
    with open(pipe_file_path) as fifo:
        while True:
            try:
                print("start read pipe data")
                data = fifo.read()
                if len(data) == 0:
                    print("writer close")
                    break 
                print(data)
            except Exception as e:
                print(e)

在控制台启动：

[root@ts]# python test.py
open fifo pipe

另起一个控制台查看该管道文件：

[root@ts]# ll /tmp/fifo_pipe
prw-r----- 1 root root 0 May 16 10:43 /tmp/fifo_pipe

说明FIFO管道确实是以文件的形式存在的

执行如下命令向管道发送消息：

[root@ts]# echo "ts" > /tmp/fifo_pipe

此时可以看到test.py的输出如下：

[root@ts]# python test.py
open fifo pipe
start read pipe data
ts

start read pipe data
writer close

三、消息队列

消息队列，是消息的链接表，存放在内核中。一个消息队列由一个标识符（即队列ID）来标识。

1、特征

  • 消息队列是面向记录的，其中的消息具有特定的格式以及特定的优先级。
  • 消息队列独立于发送与接收进程。进程终止时，消息队列及其内容并不会被删除。
  • 消息队列可以实现消息的随机查询,消息不一定要以先进先出的次序读取,也可以按消息的类型读取。

2、原型

#include <sys/msg.h>
// 创建或打开消息队列：成功返回队列ID，失败返回-1
int msgget(key_t key, int flag);
// 添加消息：成功返回0，失败返回-1
int msgsnd(int msqid, const void *ptr, size_t size, int flag);
// 读取消息：成功返回消息数据的长度，失败返回-1
int msgrcv(int msqid, void *ptr, size_t size, long type,int flag);
// 控制消息队列：成功返回0，失败返回-1
int msgctl(int msqid, int cmd, struct msqid_ds *buf);

在以下两种情况下，msgget将创建一个新的消息队列：

如果没有与键值key相对应的消息队列，并且flag中包含了IPC_CREAT标志位。 key参数为IPC_PRIVATE。 函数msgrcv在读取消息队列时，type参数有下面几种情况：

  • type == 0，返回队列中的第一个消息；
  • type > 0，返回队列中消息类型为 type 的第一个消息；
  • type < 0，返回队列中消息类型值小于或等于 type 绝对值的消息，如果有多个，则取类型值最小的消息。

可以看出，type值非 0 时用于以非先进先出次序读消息。也可以把 type 看做优先级的权值。（其他的参数解释，请自行Google之）

3、示例

# coding=utf-8
from multiprocessing import Queue, Process
import time, random, os


def write(que):
    for value in range(10):
        print('put %s to queue...' % value)
        # 1、如果block使用默认值，且没有设置timeout（单位秒）， 
        # 消息列队如果已经没有空间可写入，此时程序将被阻塞（停在写⼊状态），直到从消息列队腾出空间为止。 
        # 如果设置了timeout，则会等待timeout秒，若还没空间，则抛出”Queue.Full”异常 
        # 2、如果block值为False，消息列队如果没有空间可写入则会立刻抛出”Queue.Full”异常
        que.put(value)
        time.sleep(random.random())


def read(que):
    while True:
        if not que.empty():
            # 1、如果block使用默认值，且没有设置timeout（单位秒）， 
            # 消息列队如果为空，此时程序将被阻塞（停在读取状态），直到从消息列队读到消息为止， 
            # 如果设置了timeout，则会等待timeout秒，若还没读取到任何消息，则抛出”Queue.Empty”异常 
            # 2、如果block值为False，消息列队如果为空，则会立刻抛出”Queue.Empty”异常
            value = que.get(True)
            print('get %s from queue...' % value)
        else:
            print("wait queue data")
        time.sleep(random.random())


if __name__ == "__main__":
    que = Queue()
    qw = Process(target=write, args=(que,))
    qr = Process(target=read, args=(que,))

    qw.start()
    qr.start()

    qw.join()
    qr.join()

四、信号量

信号量（semaphore）与已经介绍过的 IPC 结构不同，它是一个计数器。信号量用于实现进程间的互斥与同步，而不是用于存储进程间通信数据。

1、特征

  • 信号量用于进程间同步，若要在进程间传递数据需要结合共享内存。
  • 信号量基于操作系统的 PV 操作，程序对信号量的操作都是原子操作。
  • 每次对信号量的 PV 操作不仅限于对信号量值加 1 或减 1，而且可以加减任意正整数。
  • 支持信号量组。

2、原型

最简单的信号量是只能取 0 和 1 的变量，这也是信号量最常见的一种形式，叫做二值信号量（Binary Semaphore）。而可以取多个正整数的信号量被称为通用信号量。 Linux 下的信号量函数都是在通用的信号量数组上进行操作，而不是在一个单一的二值信号量上进行操作。

#include <sys/sem.h>
// 创建或获取一个信号量组：若成功返回信号量集ID，失败返回-1
int semget(key_t key, int num_sems, int sem_flags);
// 对信号量组进行操作，改变信号量的值：成功返回0，失败返回-1
int semop(int semid, struct sembuf semoparray[], size_t numops);  
// 控制信号量的相关信息
int semctl(int semid, int sem_num, int cmd, ...);

当semget创建新的信号量集合时，必须指定集合中信号量的个数（即num_sems），通常为1；如果是引用一个现有的集合，则将num_sems指定为 0 。

3、示例

from multiprocessing import Process
from multiprocessing import Semaphore
from time import sleep


def sub_process(sem):
	print("enter sub process")
    sem.acquire()
    print("acquire sem")
    sleep(2)
    sem.release()
    print("release sem")


if __name__ == '__main__':
    sem = Semaphore(3)
    for i in range(5):
        sub_p = Process(target=sub_process, args=(sem,))
        sub_p.start()

执行结果如下：

enter sub process
acquire sem
enter sub process
acquire sem
enter sub process
acquire sem
enter sub process
enter sub process
release sem
acquire sem
release sem
acquire sem

release sem
release sem
release sem

可以看出，当信号量已经达到最大值，子进程会等待获取到可用的信号量时，才会执行接下来的任务

五、共享内存

共享内存（Shared Memory），指两个或多个进程共享一个给定的存储区。 操作系统负责将同一份物理地址的内存映射到多个进程的不同的虚拟地址空间中。进而每个进程都可以操作这份内存

采用共享内存通信的一个显而易见的好处效率高，因为进程可以直接读写内存，而不需要任何数据的复制。对于向管道和消息队列等通信等方式，则需要在内核和用户空间进行四次的数据复制，而共享内存则只需要两次数据复制：一次从输入文件到共享内存区，另一个从共享内存区到输出文件。 实际上，进程之间在共享内存时，并不总是读写少量数据后就解除映射，有新的通信时，再重新建立共享内存区域。而是保持共享区域，直到通信完毕为止，这样，数据内容就一直保存在共享内存中，并没有写回文件。共享内存中的内容往往是在解除映射时才写回文件的。因此，采用共享内存的通信方式效率非常高。

1、特征

  • 共享内存是最快的一种 IPC，因为进程是直接对内存进行存取。
  • 因为多个进程可以同时操作，所以需要进行同步。
  • 信号量+共享内存通常结合在一起使用，信号量用来同步对共享内存的访问。

2、原型

#include <sys/shm.h>
// 创建或获取一个共享内存：成功返回共享内存ID，失败返回-1
int shmget(key_t key, size_t size, int flag);
// 连接共享内存到当前进程的地址空间：成功返回指向共享内存的指针，失败返回-1
void *shmat(int shm_id, const void *addr, int flag);
// 断开与共享内存的连接：成功返回0，失败返回-1
int shmdt(void *addr); 
// 控制共享内存的相关信息：成功返回0，失败返回-1
int shmctl(int shm_id, int cmd, struct shmid_ds *buf);

当用shmget函数创建一段共享内存时，必须指定其 size；而如果引用一个已存在的共享内存，则将 size 指定为0 。

当一段共享内存被创建以后，它并不能被任何进程访问。必须使用shmat函数连接该共享内存到当前进程的地址空间，连接成功后把共享内存区对象映射到调用进程的地址空间，随后可像本地空间一样访问。

shmdt函数是用来断开shmat建立的连接的。 注意: 这并不是从系统中删除该共享内存，只是当前进程不能再访问该共享内存而已

shmctl函数可以对共享内存执行多种操作，根据参数 cmd 执行相应的操作。常用的是IPC_RMID（从系统中删除该共享内存）。

3、示例

下面使用python的mmap方法进行示例如下：

import contextlib
import mmap
import time
from multiprocessing import Process

with open("test.share", "w") as f:
    f.write('\x00' * 1024)


def write():
    with open('test.share', 'r+') as f:
        with contextlib.closing(
                mmap.mmap(f.fileno(), 1024, access=mmap.ACCESS_WRITE)) as m:
            m.seek(0)
            for i in range(1, 11):
                s = "msg " + str(i) + "\n" 
                print("write " + s)
                m.write(s)
                m.flush()
                time.sleep(1)


def read():
    while True:
        with open('test.share', 'r') as f:
            with contextlib.closing(
                    mmap.mmap(f.fileno(), 1024, access=mmap.ACCESS_READ)) as m:
                s = m.read(1024)
                print("read " + s)
        time.sleep(1)


if __name__ == '__main__':
    w = Process(target=write)
    r = Process(target=read)
    w.start()
    r.start()
    w.join()
    r.join()

执行结果如下：

[root@ts]# python2 test.py
write msg 1
read msg 1
read msg 1
write msg 2
read msg 1 msg 2
write msg 3
read msg 1 msg 2 msg 3
write msg 4
read msg 1 msg 2 msg 3 msg 4
write msg 5
read msg 1 msg 2 msg 3 msg 4 msg 5
write msg 6
read msg 1 msg 2 msg 3 msg 4 msg 5 msg 6
write msg 7
read msg 1 msg 2 msg 3 msg 4 msg 5 msg 6 msg 7
write msg 8
read msg 1 msg 2 msg 3 msg 4 msg 5 msg 6 msg 7 msg 8
write msg 9
read msg 1 msg 2 msg 3 msg 4 msg 5 msg 6 msg 7 msg 8 msg 9
write msg 10
read msg 1 msg 2 msg 3 msg 4 msg 5 msg 6 msg 7 msg 8 msg 9 msg 10

注：Python 3.8 新增 multiprocessing.SharedMemory 支持共享内存，直接操作 multiprocessing.SharedMemory 会产生数据竞争，不应该直接使用，应该使用 multiprocessing.Value 和 multiprocessing.Array 这种更高层的抽象

套接字

以太网套接字

也就是我们跨网络使用的tcp/udp

Unix域套接字

当同一个机器的多个进程使用普通套接字进行通信时，需要经过网络协议栈，这非常浪费，因为同一个机器根本没有必要走网络。所以Unix提供了一个套接字的特殊版本，它使用和套接字一摸一样的api，但是地址不再是网络端口，而是文件。相当于我们通过某个特殊文件来进行套接字通信。

无名套接字socketpair

Unix系统提供了无名套接字socketpair，不需要端口也可以创建套接字，父子进程通过socketpair来进行全双工通信。 跟unix域套接字的区别是，不需要创建socket文件并绑定监听。 socketpair返回两个套接字对象，一个用于读一个用于写，它有点类似于pipe，只不过pipe返回的是两个文件描述符，都是整数。所以写起代码形式上跟pipe几乎没有什么区别。

6 个最好的 Python IDE 和代码编辑器
上一篇
Python中的GIL机制详解
下一篇