美文网首页
python基础进阶:进程间通信,通信之间相互得关系

python基础进阶:进程间通信,通信之间相互得关系

作者: 孤城暮雨丶 | 来源:发表于2021-01-07 14:14 被阅读0次

    本文的文字及图片来源于网络,仅供学习、交流使用,不具有任何商业用途,版权归原作者所有,如有问题请及时联系我们以作处理

    本文章来自腾讯云 作者:我是李超人

    私信小编回复01可领取学习资料以及学习视频

    程序在创建子进程时,会完全复制一份主进程的环境,包括变量,函数,类等。所以在子进程中使用的变量,函数,类和主进程之间隔离的,子进程之间也是隔离的。
    看下面这个案例:

    from multiprocessing import Process
    
    AGE = 1
    
    def hello():
        print('hello')
    
    def greet(names):
        global AGE
        AGE += 1
        names.append('jack')
        print('======子进程代码=======')
        print('AGE = %s, id = %s'%(AGE, id(AGE)))
        print('names = %s , id = %s'%(names, id(names)))
        print('hello id = %s'%(id(hello)))
        print('======子进程代码=======')
    
    if __name__ == '__main__':
        names = ['peter']
        p = Process(target=greet, args=(names,))
        p.start()
        p.join()
    
        print('第二次执行子进程')
        p2 = Process(target=greet, args=(names,))
        p2.start()
        p2.join()
    
        print('======父进程代码=======')
        print('AGE = %s, id = %s'%(AGE, id(AGE)))
        print('names = %s , id = %s'%(names, id(names)))
        print('hello id = %s' % (id(hello)))
        print('======父进程代码=======')
    输出:
    ======子进程代码=======
    AGE = 2, id = 140736619595456
    names = ['peter', 'jack'] , id = 2672604171264
    hello id = 2672609140448
    ======子进程代码=======
    第二次执行子进程
    ======子进程代码=======
    AGE = 2, id = 140736619595456
    names = ['peter', 'jack'] , id = 1613780831232
    hello id = 1613785865952
    ======子进程代码=======
    ======父进程代码=======
    AGE = 1, id = 140736619595424
    names = ['peter'] , id = 2006331253248
    hello id = 2006329121232
    ======父进程代码=======
    
    

    从案例中可以看到,进程把所有的变量,函数都新复制了一份,即使是全局变量进程之间也是隔离的。所以进程之间要想共享数据需要使用进程间的通信,进程间通行有两种方式,第一种是管道(Pipe),第二种是队列。

    Pipe

    pipe常用于两个进程,两个进程分别位于管道的两端
    Pipe方法返回管道的两端conn1和conn2,Pipe有一个duplex参数,默认为True,即全双工模式,若为false,conn1只能接收消息,conn2只能发送消息。

    class MyProcess(Process):
    
        def __init__(self, threadname, conn):
            super().__init__()
            self.threadname = threadname
            self.conn = conn
    
        def run(self):
            self.conn.send('我是' + self.threadname)
            print(self.threadname + '-接受消息:' + self.conn.recv())
            self.conn.close()
    
    if __name__ == '__main__':
        # 建立管道,拿到管道的两端,双工通信方式,两端都可以收发消息
        conn1, conn2 = Pipe()
        myProcess1 = MyProcess('thread1', conn1)
        myProcess2 = MyProcess('thread2', conn2)
    
        myProcess1.start()
        myProcess2.start()
    输出:
    thread2-接受消息:我是thread1
    thread1-接受消息:我是thread2
    
    

    Queue

    Queue的一些常用方法的:
    Queue(n):初始化一个消息队列,并指定这个队列中最多能够容纳多少条消息。
    put(obj,[block[,timeout]]):推入一条消息到这个队列中。默认是阻塞的,也就是说如果这个消息队列中已经满了,那么会会一直等待,将这个消息添加到消息队列中。timeout可以指定这个阻塞最长的时间,如果超过这个时间还是满的,就会抛出异常。
    put_nowait() :非阻塞的推入一条消息,如果这个队列已经满了,那么会立马抛出异常。
    qsize():获取这个消息队列消息的数量。
    full():判断这个消息队列是否满了。
    empty():判断这个消息队列是否空了。
    get([block[,timeout]]):获取队列中的一条消息,然后将其从队列中移除,block默认为True。如果设置block为False,那么如果没值,会立马抛出异常。timeout指定如果多久没有获取到值后会抛出异常。

    看个栗子:

    from multiprocessing import Queue
    
    # 初始化一个Queue对象,最多只能存放三条消息
    q = Queue(3)
    
    # 存放第一条消息
    q.put('m1')
    # 存放第二条消息
    q.put('m2')
    
    # 判断这个队列中是否已经满了
    print(q.full())
    
    # 存放第三条消息
    q.put('m3')
    
    # 判断这个队列中的消息是否已经满了
    print(q.full())
    
    # 因为如果消息队列已经满了,那么再put进去的时候就会报错
    try:
        # q.put('m4', block=False)# 队列满了立即抛出异常
        # q.put('m4',block=True, timeout=2) # 可以被阻塞,等待时间超过两秒抛出异常
        q.put_nowait('m4')
    except:
        print('消息队列已经满了,现有消息数量:%s' % q.qsize())
    
    print(q.get())
    
    输出:
    False
    True
    消息队列已经满了,现有消息数量:3
    m1
    
    

    利用queue作为进程间通信来实现生产者和消费者

    from multiprocessing import Process, Queue
    import os
    
    def write(q):
        for x in ['m1', 'm2', 'm3']:
            q.put(x)
            print('子进程已经存放了消息%s, id : %s' % (x, os.getpid()))
    
    def read(q):
        while True:
            try:
                msg = q.get(block=True, timeout=3)
                print('子进程已经读出了消息%s, id : %s' % (msg, os.getpid()))
            except:
                print('所有消息已经读完了')
                break
    
    if __name__ == '__main__':
        q = Queue()
        pw = Process(target=write, args=(q,))
        pr = Process(target=read, args=(q,))
    
        pw.start()
        pr.start()
        pw.join()
    输出:
    子进程已经存放了消息m1, id : 28740
    子进程已经存放了消息m2, id : 28740
    子进程已经读出了消息m1, id : 17936
    子进程已经读出了消息m2, id : 17936
    所有消息已经读完了
    
    

    进程池间通信

    进程池间的通信使用Manager().Queue(),不能使用Queue(会报错,Queue objects should only be shared between processes through inheritance),Manager().Queue()和Queue的使用方法是一样的。

    from multiprocessing import Pool, Manager
    import os
    
    def write(q):
        for x in ['m1', 'm2']:
            q.put(x)
            print('子进程已经存放了消息%s, id : %s' % (x, os.getpid()))
    
    def read(q):
        while True:
            try:
                msg = q.get(block=True, timeout=3)
                print('子进程已经读出了消息%s, id : %s' % (msg, os.getpid()))
            except:
                print('所有消息已经读完了')
                break
    
    if __name__ == '__main__':
        q = Manager().Queue()
        pool = Pool(2)
        pool.apply(write, args=(q,))
        pool.apply(read, args=(q,))
    
        pool.close()
        pool.join()
    输出:
    子进程已经存放了消息m1, id : 3052
    子进程已经存放了消息m2, id : 3052
    子进程已经读出了消息m1, id : 22556
    子进程已经读出了消息m2, id : 22556
    所有消息已经读完了
    
    

    做一下小总结:Python进程间数据是不共享的,所有的函数,变量,类都会被重新复制一份,要想让进程间可以共享数据,需要用到进程通信技术。比如pipe,和queue。Pipe常用于两个进程间的两端通信,实际用得较少。通常都是通过queue来实现进程间通信,进程池间通行是通过Manager.Queue。

    相关文章

      网友评论

          本文标题:python基础进阶:进程间通信,通信之间相互得关系

          本文链接:https://www.haomeiwen.com/subject/ahjzoktx.html