专栏首页大数据入坑指南Python自学成才之路 进程间通信

Python自学成才之路 进程间通信

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

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

  1. pipe常用于两个进程,两个进程分别位于管道的两端
  2. 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。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Python自学成才之路 多进程开发

    在python中可以通过os.fork()创建子进程,但是这种方式只能在linux,unix,mac下面使用,不能跨平台,所以一般不推荐使用这种方式。Pytho...

    我是李超人
  • Python 处理时间差

    1.计算两个时间相差多少天,或者几个小时,可以使用datetime模块解决这个问题

    我是李超人
  • Python 输入时间字符串以分钟单位计算时间差

    之前转载过一篇使用python dateTime模块处理时间差的文章,文章中讲解了如何按照days,hours和seconds单位来计算时间差。这里讲解一下如何...

    我是李超人
  • Mysql性能调优(五)

      上篇文章我们介绍了MySQL体系结构以及存储引擎的相关介绍。接下来,我们给大家介绍SQL语句的优化以及具体的sql优化步骤。接下来我们介绍优化sql的具体步...

    stefan666
  • 面试官常问的Spring依赖注入和Bean的装配问题,今天给大家讲清楚!

    小伙伴儿们,Spring 的依赖注入以及 Bean 的装配是面试常问的知识点,今天我们来学习一下Spring中的依赖注入方式,以及如何将自己开发的Bean装配到...

    用户7544680
  • 面试官问:MySQL 的自增 ID 用完了,怎么办?

    公众号改版后文章乱序推荐,希望你可以点击上方“Java进阶架构师”,点击右上角,将我们设为★“星标”!这样才不会错过每日进阶架构文章呀。

    java进阶架构师
  • 程序猿的日常——Mybatis现学现卖

    最近有一个小项目需求,需要用spring mvc + mybatis实现一个复杂的配置系统。其中遇到了很多不太常见的问题,在这里特意记录下: 主要涉及的内容有...

    用户1154259
  • 操作系统 进程线程模型 进程控制块及进程控制

    调度信息供进程使用时使用,描述了进程当前所处的状况,他包括进程名、存储信息、进程号、优先级、当前状态、资源清单、“家族”关系、消息队列指针、进程队列指针和当前打...

    Debug客栈
  • MYSQL回顾(表关系相关)

    比如有两个表,分别是书籍表和出版社表。书籍和出版社是典型的多对一关系,即一本书只能由一个出版社出版,一个出版社可以出版多本书。则书籍表应该有一个外键press_...

    VV木公子
  • SpringFramework之RequestBodyAdvice

    本来RequestBodyAdvice和ResponseBodyAdvice是成对一起的,这里先分析RequestBodyAdvice.

    克虏伯

扫码关注云+社区

领取腾讯云代金券