多进程通信实例分析

操作系统会给每一个创建而成的进程赋予一个独立的地址空间, 不同进程对应的地址空间是全然隔离的, 所以要是不附加其他措施, 它们压根感觉不到彼此的存在。那么进程之间究竟要怎样去完成通信? 它们之间的关联到底是怎样的? 其实现原理又是什么? 本文于是借助相对简单的方式来聊一下进程之间的通信一事。依旧是那句表述, 原理是相同的, 期望能够透过具体的例子去体会一下里面本质性的内容。

向下, 尽可能以简易的形式, 去分别介绍一下每一种通信方式, 具体的详细内容能够参照文档来加以使用。

管道

先瞧一瞧那最为简单、年代久远有着的一种 IPC, 它是管道。一般所讲的是无名管道, 从本质层面而言可以视作一种文件, 仅仅存在于内存里, 不会进行存盘。不同的进程借助系统所给予的接口去朝着管道之中读取或者写入数据。

也就意味着, 我们借助这样一种中间介质, 来给进程提供交流途径。无名管道的局限性在于, 通常仅用于存在直接关联关系的父子进程。下面通过一个简单事例, 来看一下它的用法。

复制代码

from , Pipe

def (pname, conn):

conn.send("Data@subprocess")
print(conn.recv())          # Data@parentprocess

if name == '':

conn1, conn2 = Pipe(True)
sub_proc = Process(target=pstart, args=('subprocess', conn2))
sub_proc.start()
print (conn1.recv())        # Data@subprocess
conn1.send("Data@parentprocess")
sub_proc.join()

复制代码

管道通信三步曲：

创建Pipe，得到两个对象conn1和conn2；

父进程持有conn1，将conn2传递给子进程；

父进程与子进程, 凭借对所持有对象开展send操作以及recv操作, 从而达成数据传递以及接受的目的。

前边我们所构建的是全双工管道, 同样能够构建半双工管道, 其具体的运用能够参照官网的描述。

具名管道（FIFO）

呈现在上面所介绍的那些管道, 其主要用途是针对存在直接关联关系的进程, 然而, 该管道的局限性相对而言是比较大的。接下来, 让我们一同去瞧一瞧那种能够在任意进程之间展开通信的具名管道。

是因为平台之上os模块不存在属性呀, 所以呢这个例子仅仅能够在linux上去运行（测试环境为7, 版本是2.7.5）。

复制代码

!/usr/bin/

os, time

from

= "./pipe.in"

= "./pipe.out"

def ():

if os.path.exists(input_pipe):
    os.remove(input_pipe)
if os.path.exists(output_pipe):
    os.remove(output_pipe)
os.mkfifo(output_pipe)
os.mkfifo(input_pipe)
in1 = os.open(input_pipe, os.O_RDONLY)        # read from pipe.in
out1 = os.open(output_pipe, os.O_SYNC | os.O_CREAT | os.O_RDWR)
while True:
    read_data = os.read(in1, 1024)
    print("received data from pipe.in: %s @consumer" % read_data)
    if len(read_data) == 0:
        time.sleep(1)
        continue
    if "exit" in read_data:
        break
    os.write(out1, read_data)
os.close(in1)
os.close(out1)

def ():

in2 = None
out2 = os.open(input_pipe, os.O_SYNC | os.O_CREAT | os.O_RDWR)
for i in range(1, 4):
    msg = "msg " + str(i)
    len_send = os.write(out2, msg)
    print("------product msg: %s by producer------" % msg)
    if in2 is None:
        in2 = os.open(output_pipe, os.O_RDONLY)        # read from pipe.out
    data = os.read(in2, 1024)
    if len(data) == 0:
        break
    print("received data from pipe.out: %s @producer" % data)
    time.sleep(1)
os.write(out2, 'exit')
os.close(in2)
os.close(out2)

if name == '':

pconsumer = Process(target=consumer, args=())
pproducer = Process(target=producer, args=())
pconsumer.start()
time.sleep(0.5)
pproducer.start()
pconsumer.join()
pproducer.join()

复制代码

运行流程如下：

每一轮的过程如下：

进程往pipe.in文件中写入消息数据；

进程从pipe.in文件中读入消息数据；

进程往pipe.out文件中写入回执消息数据；

进程从pipe.out文件中读出回执消息数据；

结果如下：

View Code

存在两个进程, 它们之间不存在直接的关系。其中每个进程, 都分别有一个用于读文件的操作, 以及一个用于写文件的操作。假若是这样的情况, 若两个进程的读写文件存在关联, 那么它们便能够进行通信。

消息队列（Queue）

往队列里添加数据, 或者从队列里获取数据, 进程借此来传递消息数据。这是个简单的例子。

复制代码

from , Queue

time

def (que):

for product in ('Orange', 'Apple', ''):
    print('put product: %s to queue' % product)
    que.put(product)
    time.sleep(0.5)
    res = que.get()
    print('consumer result: %s' % res)

def (que):

while True:
    product = que.get()
    print('get product:%s from queue' % product)
    que.put('suc!')
    time.sleep(0.5)
    if not product:
        break

if name == '':

que = Queue(1)
p = Process(target=producer, args=(que))
c = Process(target=consumer, args=(que,))
p.start()
c.start()
p.join()
c.join()

复制代码

这个例子比较简单，queue的具体用法可以参考一下官网。

结果：

View Code

这里有几点需要注意下：

能够对队列的容量予以指定, 要是出现超出所指定容量的情况, 便会产生异常, 这种异常为: raise Full。

默认put和get均会阻塞当前进程；

设若put未被设定成处于阻塞状态, 那么存在这样一种可能性, 即自身从中队列里取出由自身放置进去的数据。

共享内存

共享内存, 是一种常用的通信方式, 用于进程之间, 它具备高效的特点, 而为了确保共享内存能够被有序访问, 针对进程方面, 有必要采取额外的同步措施。

下面的这一个例子, 只是单纯简单地演示了, 在其中怎样于不同进程之间, 运用共享内存去进行通信呢。

复制代码

from

mmap

time

def ():

with contextlib.closing(mmap.mmap(-1, 1024, tagname='cnblogs', access=mmap.ACCESS_WRITE)) as mem:
    for share_data in ("Hello", "Alpha_Panda"):
        mem.seek(0)
        print('Write data:== %s == to share memory!' % share_data)
        mem.write(str.encode(share_data))
        mem.flush()
        time.sleep(0.5)

def ():

while True:
    invalid_byte, empty_byte = str.encode('\x00'), str.encode('')
    with contextlib.closing(mmap.mmap(-1, 1024, tagname='cnblogs', access=mmap.ACCESS_READ)) as mem:
        share_data = mem.read(1024).replace(invalid_byte, empty_byte)
        if not share_data:
            """ 当共享内存没有有效数据时结束reader """
            break
        print("Get data:== %s == from share memory!" % share_data.decode())
    time.sleep(0.5)

if name == '':

p_reader = Process(target=reader, args=())
p_writer = Process(target=writer, args=())
p_writer.start()
p_reader.start()
p_writer.join()
p_reader.join()

复制代码

执行结果：

写数据, 等于, 向分享, 问好, 等于, 要分享！

Write data:== == to share !

获取数据, 等于, 来自共享的, “你好”！

Get data:== == from share !

下面简单的来说明一下共享内存的原理；

进程虚拟地址到物理地址的一个映射关如下：

上面这个图已经很明白的展示了共享内存的原理。

其一, 于正常状况之下, 不同进程的线性地址空间会被映射至各异的物理内存页, 其二, 如此方式之下, 无论其他进程怎样对物理内存予以修改, 其三, 皆不会对其他进程产生影响。

在进程共享内存情形下, 右边所呈现的是, 不同进程的部分线性地址会被映射至同一物理页, 而一旦有一个进程针对此物理页作出修改, 那么对于另一个进程而言将会立即可见。

之所以说潜在的问题即必然要采取进程同步措施, 是因为此对于共享内存的访问是必须达成互斥状态的结果。这一点能够通过借助信号量这一方式来予以实现。

通信

最后再来介绍一种可以跨主机的进程间通信：。

对于那些知晓网络编程的人而言, 针对这个理应都相对熟悉, 不但能够在不同主机之间开展通信, 甚至某些时候能够运用于同一主机里不同进程之间来进行通信。

就这一部分代码而言, 都是相对简单且常见的, 在此处只是单纯运用流程图去把通信的流程以及相关接口给呈现出来。

某个进程、在客户端之上、使用着, 服务器之上、监听程序、与之进行通信, 这样的一个流程、被上图所表示。

小结

这里针对常见的进程间通信相关概念, 进行了简单介绍, 这里针对常见的进程间通信相关实例, 也进行了简单介绍。期望本文能够使得你对于进程间通信拥有一个更为深入的理解, 期望本文能够使得你对于进程间通信拥有一个更为深入的认识。

结合之前几篇, 有对线程概念的介绍, 有对进程概念的介绍, 还有对线程间同步一些措施的介绍, 相信应该对线程和进程相关概念, 有一个简单清晰的认识了。

原文地址