Python进程间通信

原文链接：https://zhuanlan.zhihu.com/p/37370601

作者：老钱

其他参考：https://www.cnblogs.com/zgq0/p/8780893.html

进程间通信的几种主要手段简介：

1. 管道（Pipe）及有名管道（named pipe）：管道可用于具有亲缘关系进程间的通信，有名管道克服了管道没有名字的限制，因此，除具有管道所具有的功能外，它还允许无亲缘关系进程间的通信；
2. 信号（Signal）：信号是比较复杂的通信方式，用于通知接受进程有某种事件发生，除了用于进程间通信外，进程还可以发送信号给进程本身；linux除了支持Unix早期信号语义函数sigal外，还支持语义符合Posix.1标准的信号函数sigaction（实际上，该函数是基于BSD的，BSD为了实现可靠信号机制，又能够统一对外接口，用sigaction函数重新实现了signal函数）；
3. 报文（Message）队列（消息队列）：消息队列是消息的链接表，包括Posix消息队列system V消息队列。有足够权限的进程可以向队列中添加消息，被赋予读权限的进程则可以读走队列中的消息。消息队列克服了信号承载信息量少，管道只能承载无格式字节流以及缓冲区大小受限等缺点。
4. 共享内存：使得多个进程可以访问同一块内存空间，是最快的可用IPC形式。是针对其他通信机制运行效率较低而设计的。往往与其它通信机制，如信号量结合使用，来达到进程间的同步及互斥。
5. 信号量（semaphore）：主要作为进程间以及同一进程不同线程之间的同步手段。
6. 套接口（Socket）：更为一般的进程间通信机制，可用于不同机器之间的进程间通信。起初是由Unix系统的BSD分支开发出来的，但现在一般可以移植到其它类Unix系统上：Linux和System V的变种都支持套接字。

文件

使用文件进行通信是最简单的一种通信方式，子进程将结果输出到临时文件，父进程从文件中读出来。文件名使用子进程的进程id来命名。进程随时都可以通过os.getpid()来获取自己的进程id。

# coding: utf-8

import os
import sys
import math


def slice(mink, maxk):
    s = 0.0
    for k in range(mink, maxk):
        s += 1.0/(2*k+1)/(2*k+1)
    return s


def pi(n):
    pids = []
    unit = n / 10
    for i in range(10):  # 分10个子进程
        mink = unit * i
        maxk = mink + unit
        pid = os.fork()
        if pid > 0:
            pids.append(pid)
        else:
            s = slice(mink, maxk)  # 子进程开始计算
            with open("%d" % os.getpid(), "w") as f:
                f.write(str(s))
            sys.exit(0)  # 子进程结束
    sums = []
    for pid in pids:
        os.waitpid(pid, 0)  # 等待子进程结束
        with open("%d" % pid, "r") as f:
            sums.append(float(f.read()))
        os.remove("%d" % pid)  # 删除通信的文件
    return math.sqrt(sum(sums) * 8)


print pi(10000000)

输出

3.14159262176

管道pipe

管道是Unix进程间通信最常用的方法之一，它通过在父子进程之间开通读写通道来进行双工交流。我们通过os.read()和os.write()来对文件描述符进行读写操作，使用os.close()关闭描述符。

上图为单进程的管道

 

上图为父子进程分离后的管道

# coding: utf-8

import os
import sys
import math


def slice(mink, maxk):
    s = 0.0
    for k in range(mink, maxk):
        s += 1.0/(2*k+1)/(2*k+1)
    return s


def pi(n):
    childs = {}
    unit = n / 10
    for i in range(10):  # 分10个子进程
        mink = unit * i
        maxk = mink + unit
        r, w = os.pipe()
        pid = os.fork()
        if pid > 0:
            childs[pid] = r  # 将子进程的pid和读描述符存起来
            os.close(w)  # 父进程关闭写描述符，只读
        else:
            os.close(r)  # 子进程关闭读描述符，只写
            s = slice(mink, maxk)  # 子进程开始计算
            os.write(w, str(s))
            os.close(w)  # 写完了，关闭写描述符
            sys.exit(0)  # 子进程结束
    sums = []
    for pid, r in childs.items():
        sums.append(float(os.read(r, 1024)))
        os.close(r)  # 读完了，关闭读描述符
        os.waitpid(pid, 0)  # 等待子进程结束
    return math.sqrt(sum(sums) * 8)


print pi(10000000)

输出

3.14159262176

有名管道fifo

相对于管道只能用于父子进程之间通信，Unix还提供了有名管道可以让任意进程进行通信。有名管道又称fifo，它会将自己注册到文件系统里一个文件，参数通信的进程通过读写这个文件进行通信。
fifo要求读写双方必须同时打开才可以继续进行读写操作，否则打开操作会堵塞直到对方也打开。

# coding: utf-8

import os
import sys
import math


def slice(mink, maxk):
    s = 0.0
    for k in range(mink, maxk):
        s += 1.0/(2*k+1)/(2*k+1)
    return s


def pi(n):
    childs = []
    unit = n / 10
    fifo_path = "/tmp/fifo_pi"
    os.mkfifo(fifo_path)  # 创建named pipe
    for i in range(10):  # 分10个子进程
        mink = unit * i
        maxk = mink + unit
        pid = os.fork()
        if pid > 0:
            childs.append(pid)
        else:
            s = slice(mink, maxk)  # 子进程开始计算
            with open(fifo_path, "w") as ff:
                ff.write(str(s) + "
")
            sys.exit(0)  # 子进程结束
    sums = []
    while True:
        with open(fifo_path, "r") as ff:
            # 子进程关闭写端，读进程会收到eof
            # 所以必须循环打开，多次读取
            # 读够数量了就可以结束循环了
            sums.extend([float(x) for x in ff.read(1024).strip().split("
")])
            if len(sums) == len(childs):
                break
    for pid in childs:
        os.waitpid(pid, 0)  # 等待子进程结束
    os.unlink(fifo_path)  # 移除named pipe
    return math.sqrt(sum(sums) * 8)


print pi(10000000)

输出

3.14159262176

以太网套接字

套接字无疑是通信使用最为广泛的方式了，它不但能跨进程还能跨网络。今天英特网能发达成这样，全拜套接字所赐。不过作为同一个机器的多进程通信还是挺浪费的。暂不讨论这个，还是先看看它如何使用吧。

# coding: utf-8

import os
import sys
import math
import socket


def slice(mink, maxk):
    s = 0.0
    for k in range(mink, maxk):
        s += 1.0/(2*k+1)/(2*k+1)
    return s


def pi(n):
    childs = []
    unit = n / 10
    servsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 注意这里的AF_INET表示普通套接字
    servsock.bind(("localhost", 0))  # 0表示随机端口
    server_address = servsock.getsockname()  # 拿到随机出来的地址，给后面的子进程使用
    servsock.listen(10)  # 监听子进程连接请求
    for i in range(10):  # 分10个子进程
        mink = unit * i
        maxk = mink + unit
        pid = os.fork()
        if pid > 0:
            childs.append(pid)
        else:
            servsock.close()  # 子进程要关闭servsock引用
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.connect(server_address)  # 连接父进程套接字
            s = slice(mink, maxk)  # 子进程开始计算
            sock.sendall(str(s))
            sock.close()  # 关闭连接
            sys.exit(0)  # 子进程结束
    sums = []
    for pid in childs:
        conn, _ = servsock.accept()  # 接收子进程连接
        sums.append(float(conn.recv(1024)))
        conn.close()  # 关闭连接
    for pid in childs:
        os.waitpid(pid, 0)  # 等待子进程结束
    servsock.close()  # 关闭套接字
    return math.sqrt(sum(sums) * 8)


print pi(10000000)

输出

3.14159262176

Unix域套接字

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

# coding: utf-8

import os
import sys
import math
import socket


def slice(mink, maxk):
    s = 0.0
    for k in range(mink, maxk):
        s += 1.0/(2*k+1)/(2*k+1)
    return s


def pi(n):
    server_address = "/tmp/pi_sock"  # 套接字对应的文件名
    childs = []
    unit = n / 10
    servsock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    servsock.bind(server_address)
    servsock.listen(10)  # 监听子进程连接请求
    for i in range(10):  # 分10个子进程
        mink = unit * i
        maxk = mink + unit
        pid = os.fork()
        if pid > 0:
            childs.append(pid)
        else:
            servsock.close()  # 子进程要关闭servsock引用
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            sock.connect(server_address)  # 连接父进程套接字
            s = slice(mink, maxk)  # 子进程开始计算
            sock.sendall(str(s))
            sock.close()  # 关闭连接
            sys.exit(0)  # 子进程结束
    sums = []
    for pid in childs:
        conn, _ = servsock.accept()  # 接收子进程连接
        sums.append(float(conn.recv(1024)))
        conn.close()  # 关闭连接
    for pid in childs:
        os.waitpid(pid, 0)  # 等待子进程结束
    servsock.close()  # 关闭套接字
    os.unlink(server_address)  # 移除套接字文件
    return math.sqrt(sum(sums) * 8)


print pi(10000000)

输出

3.14159262176

无名套接字socketpair

我们知道跨网络通信免不了要通过套接字进行通信，但是本例的多进程是在同一个机器上，用不着跨网络，使用普通套接字进行通信有点浪费。

上图为单进程的socketpair

 

上图为父子进程分离后的socketpair

为了解决这个问题，Unix系统提供了无名套接字socketpair，不需要端口也可以创建套接字，父子进程通过socketpair来进行全双工通信。

socketpair返回两个套接字对象，一个用于读一个用于写，它有点类似于pipe，只不过pipe返回的是两个文件描述符，都是整数。所以写起代码形式上跟pipe几乎没有什么区别。

我们使用sock.send()和sock.recv()来对套接字进行读写，通过sock.close()来关闭套接字对象。

# coding: utf-8

import os
import sys
import math
import socket


def slice(mink, maxk):
    s = 0.0
    for k in range(mink, maxk):
        s += 1.0/(2*k+1)/(2*k+1)
    return s


def pi(n):
    childs = {}
    unit = n / 10
    for i in range(10):  # 分10个子进程
        mink = unit * i
        maxk = mink + unit
        rsock, wsock = socket.socketpair()
        pid = os.fork()
        if pid > 0:
            childs[pid] = rsock
            wsock.close()
        else:
            rsock.close()
            s = slice(mink, maxk)  # 子进程开始计算
            wsock.send(str(s))
            wsock.close()
            sys.exit(0)  # 子进程结束
    sums = []
    for pid, rsock in childs.items():
        sums.append(float(rsock.recv(1024)))
        rsock.close()
        os.waitpid(pid, 0)  # 等待子进程结束
    return math.sqrt(sum(sums) * 8)


print pi(10000000)

输出

3.14159262176

OS消息队列

操作系统也提供了跨进程的消息队列对象可以让我们直接使用，只不过python没有默认提供包装好的api来直接使用。我们必须使用第三方扩展来完成OS消息队列通信。第三方扩展是通过使用Python包装的C实现来完成的。

OS消息队列有两种形式，一种是posix消息队列，另一种是systemv消息队列，有些操作系统两者都支持，有些只支持其中的一个，比如macos仅支持systemv消息队列，我本地的python的docker镜像是debian linux，它仅支持posix消息队列。

posix消息队列
我们先使用posix消息队列来完成圆周率的计算，posix消息队列需要提供一个唯一的名称，它必须是/开头。close()方法仅仅是减少内核消息队列对象的引用，而不是彻底关闭它。unlink()方法才能彻底销毁它。O_CREAT选项表示如果不存在就创建。向队列里塞消息使用send方法，收取消息使用receive方法，receive方法返回一个tuple，tuple的第一个值是消息的内容，第二个值是消息的优先级。之所以有优先级，是因为posix消息队列支持消息的排序，在send方法的第二个参数可以提供优先级整数值，默认为0，越大优先级越高。

# coding: utf-8

import os
import sys
import math
from posix_ipc import MessageQueue as Queue


def slice(mink, maxk):
    s = 0.0
    for k in range(mink, maxk):
        s += 1.0/(2*k+1)/(2*k+1)
    return s


def pi(n):
    pids = []
    unit = n / 10
    q = Queue("/pi", flags=os.O_CREAT)
    for i in range(10):  # 分10个子进程
        mink = unit * i
        maxk = mink + unit
        pid = os.fork()
        if pid > 0:
            pids.append(pid)
        else:
            s = slice(mink, maxk)  # 子进程开始计算
            q.send(str(s))
            q.close()
            sys.exit(0)  # 子进程结束
    sums = []
    for pid in pids:
        sums.append(float(q.receive()[0]))
        os.waitpid(pid, 0)  # 等待子进程结束
    q.close()
    q.unlink()  # 彻底销毁队列
    return math.sqrt(sum(sums) * 8)


print pi(10000000)

输出

3.14159262176

systemv消息队列
systemv消息队列和posix消息队列用起来有所不同。systemv的消息队列是以整数key作为名称，如果不指定，它就创建一个唯一的未占用的整数key。它还提供消息类型的整数参数，但是不支持消息优先级。

# coding: utf-8

import os
import sys
import math
import sysv_ipc
from sysv_ipc import MessageQueue as Queue


def slice(mink, maxk):
    s = 0.0
    for k in range(mink, maxk):
        s += 1.0/(2*k+1)/(2*k+1)
    return s


def pi(n):
    pids = []
    unit = n / 10
    q = Queue(key=None, flags=sysv_ipc.IPC_CREX)
    for i in range(10):  # 分10个子进程
        mink = unit * i
        maxk = mink + unit
        pid = os.fork()
        if pid > 0:
            pids.append(pid)
        else:
            s = slice(mink, maxk)  # 子进程开始计算
            q.send(str(s))
            sys.exit(0)  # 子进程结束
    sums = []
    for pid in pids:
        sums.append(float(q.receive()[0]))
        os.waitpid(pid, 0)  # 等待子进程结束
    q.remove()  # 销毁消息队列
    return math.sqrt(sum(sums) * 8)


print pi(10000000)

输出

3.14159262176

共享内存

共享内存也是非常常见的多进程通信方式，操作系统负责将同一份物理地址的内存映射到多个进程的不同的虚拟地址空间中。进而每个进程都可以操作这份内存。考虑到物理内存的唯一性，它属于临界区资源，需要在进程访问时搞好并发控制，比如使用信号量。我们通过一个信号量来控制所有子进程的顺序读写共享内存。我们分配一个8字节double类型的共享内存用来存储极限的和，每次从共享内存中读出来时，要使用struct进行反序列化(unpack)，将新的值写进去之前也要使用struct进行序列化(pack)。每次读写操作都需要将读写指针移动到内存开头位置(lseek)。

# coding: utf-8
import os
import sys
import math
import struct
import posix_ipc
from posix_ipc import Semaphore
from posix_ipc import SharedMemory as Memory

def slice(mink, maxk):
    s = 0.0
    for k in range(mink, maxk):
        s += 1.0/(2*k+1)/(2*k+1)
    return s

def pi(n):
    pids = []
    unit = n / 10
    sem_lock = Semaphore("/pi_sem_lock", flags=posix_ipc.O_CREX, initial_value=1)  # 使用一个信号量控制多个进程互斥访问共享内存
    memory = Memory("/pi_rw", size=8, flags=posix_ipc.O_CREX)
    os.lseek(memory.fd, 0, os.SEEK_SET)  # 初始化和为0.0的double值
    os.write(memory.fd, struct.pack('d', 0.0))
    for i in range(10):  # 分10个子进程
        mink = unit * i
        maxk = mink + unit
        pid = os.fork()
        if pid > 0:
            pids.append(pid)
        else:
            s = slice(mink, maxk)  # 子进程开始计算
            sem_lock.acquire()
            try:
                os.lseek(memory.fd, 0, os.SEEK_SET)
                bs = os.read(memory.fd, 8)  # 从共享内存读出来当前值
                cur_val, = struct.unpack('d', bs)  # 反序列化，逗号不能少
                cur_val += s  # 加上当前进程的计算结果
                bs = struct.pack('d', cur_val) # 序列化
                os.lseek(memory.fd, 0, os.SEEK_SET)
                os.write(memory.fd, bs)  # 写进共享内存
                memory.close_fd()
            finally:
                sem_lock.release()
            sys.exit(0)  # 子进程结束
    sums = []
    for pid in pids:
        os.waitpid(pid, 0)  # 等待子进程结束
    os.lseek(memory.fd, 0, os.SEEK_SET)
    bs = os.read(memory.fd, 8)  # 读出最终这结果
    sums, = struct.unpack('d', bs)  # 反序列化
    memory.close_fd()  # 关闭共享内存
    memory.unlink()  # 销毁共享内存
    sem_lock.unlink()  #  销毁信号量
    return math.sqrt(sums * 8)

print pi(10000000)

输出

3.14159262176


总结：

1.管道：速度慢，容量有限，只有父子进程能通讯    

2.FIFO：任何进程间都能通讯，但速度慢    

3.消息队列：容量受到系统限制，且要注意第一次读的时候，要考虑上一次没有读完数据的问题    

4.信号量：不能传递复杂消息，只能用来同步    

5.共享内存区：能够很容易控制容量，速度快，但要保持同步，比如一个进程在写的时候，另一个进程要注意读写的问题，相当于线程中的线程安全，当然，共享内存区同样可以用作线程间通讯，不过没这个必要，线程间本来就已经共享了同一进程内的一块内存

https://www.zhihu.com/people/codehole