管道,进程间数据共享,进程池

一:管道   (了解)

使用:from multiprocessing import Process,Pipe

知识:

1 创建管道时候:Pipe()默认是双工的,如改成False,那么conn1只能接收,conn2只能发送。

conn1,conn2=Pipe()

2 Pipe模块发送字符串不用bytes类型,直接是字符串类型。

Pipe([duplex]):在进程之间创建一条管道,并返回元组(conn1,conn2),其中conn1,
conn2表示管道两端的连接对象,强调一点:必须在产生Process对象之前产生管道.
#参数介绍:
dumplex:默认管道是全双工的,如果将duplex设置成False,conn1只能用于接收,conn2只能用于发送。
#主要方法:
conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收,recv方法会一直阻塞。
    如果连接的另外一端已经关闭,那么recv方法会抛出EOFError。
    conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象
 #其他方法:
conn1.close():关闭连接。如果conn1被垃圾回收,将自动调用此方法
conn1.fileno():返回连接使用的整数文件描述符
conn1.poll([timeout]):如果连接上的数据可用,返回True。timeout指定等待的最长时限。
如果省略此参数,方法将立即返回结果。如果将timeout射成None,操作将无限期地等待数据到达。

进程之间一个接收信息,一个发送消息

# from multiprocessing import  Process,Pipe
# def func(conn):
#     conn.send(‘hello‘)
#     conn.close()
# if __name__==‘__main__‘:#判断是否是当前文件的文件名,如果是执行下面
#     conn1,conn2=Pipe()
#     p=Process(target=func,args=(conn1,))
#     p.start()
#     print(conn2.recv())
#     # conn2.close()   注意 这个不写,我是写的错误的,这行去掉
#     p.join()

打印结果:一个hello

如果是进程之间接收多条消息的时候,不是一条,需要传递多个对象(形参是两个),执行一个时候,另一个关闭。

from multiprocessing import Process,Pipe
def func(conn1,conn2):
    conn2.close()
    while True:
        try:
            msg=conn1.recv()
            print(msg)
        except EOFError:
            conn1.close()
            break
if __name__==‘__main__‘:
    conn1,conn2=Pipe()
    Process(target=func,args=(conn1,conn2)).start()
    conn1.close()
    for i in range(20):
        conn2.send(‘hello‘)
    conn2.close()

打印结果:20个hello

pipe实现生产者消费者模型(主进程与子进程接发消息)

# pipe实现生产者消费者模型
from multiprocessing import Lock,Pipe,Process
def producer(con,pro,name,food):
    con.close()
    for i in range(100):
        f = ‘%s生产%s%s‘%(name,food,i)
        print(f)
        pro.send(f)
    pro.send(None)
    pro.send(None)
    pro.send(None)
    pro.close()

def consumer(con,pro,name,lock):
    pro.close()
    while True:
        lock.acquire()
        food = con.recv()
        lock.release()
        if food is None:
            con.close()
            break
print(‘%s吃了%s‘ % (name, food))
if __name__ == ‘__main__‘:
    con,pro = Pipe()
    lock= Lock()
    p = Process(target=producer,args=(con,pro,‘egon‘,‘泔水‘))
    c1 = Process(target=consumer, args=(con, pro, ‘alex‘,lock))
    c2 = Process(target=consumer, args=(con, pro, ‘bossjin‘,lock))
    c3 = Process(target=consumer, args=(con, pro, ‘wusir‘,lock))
    c1.start()
    c2.start()
    c3.start()
    p.start()
    con.close()
    pro.close()

注意:一定要用锁来控制。

#多个消费之之间的竞争问题带来的数据不安全问题
from multiprocessing import Process,Pipe,Lock

def consumer(produce, consume,name,lock):
    produce.close()
    while True:
        lock.acquire()
        baozi=consume.recv()
        lock.release()
        if baozi:
            print(‘%s 收到包子:%s‘ %(name,baozi))
        else:
            consume.close()
            break

def producer(produce, consume,n):
    consume.close()
    for i in range(n):
        produce.send(i)
    produce.send(None)
    produce.send(None)
    produce.close()

if __name__ == ‘__main__‘:
    produce,consume=Pipe()
    lock = Lock()
    c1=Process(target=consumer,args=(produce,consume,‘c1‘,lock))
    c2=Process(target=consumer,args=(produce,consume,‘c2‘,lock))
    p1=Process(target=producer,args=(produce,consume,30))
    c1.start()
    c2.start()
    p1.start()
    produce.close()
    consume.close()

注意:pipe 数据不安全性,一般我们用joinableQueue来实现生产模型

# pipe 数据不安全性
# IPC
# 加锁来控制操作管道的行为 来避免进程之间争抢数据造成的数据不安全现象
# 队列 进程之间数据安全的
# 管道 + 锁

进程池:重点

进程池和multiprocess.Pool模块        重点

为什么要有进程池?

在程序实际处理问题过程中,忙时会有成千上万的任务需要被执行,闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候,我们就需要去创建成千上万个进程么?首先,创建进程需要消耗时间,销毁进程也需要消耗时间。第二即便开启了成千上万的进程,操作系统也不能让他们同时执行,这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。那么我们要怎么做呢?

在这里,要给大家介绍一个进程池的概念,定义一个池子,在里面放上固定数量的进程,有需求来了,就拿一个池中的进程来处理任务,等到处理完毕,进程并不关闭,而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行,池中的进程数量不够,任务就要等待之前的进程执行任务完毕归来,拿到空闲进程才能继续执行。也就是说,池中进程的数量是固定的,那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度,还节省了开闭进程的时间,也一定程度上能够实现并发效果。

简而言之:创建一个进程池处理多个任务,当进程池有任务时候,分批处理(比如进程池设置为5,每次处理5个任务,只有当任务处理完毕才能执行下一个任务)。池中进程的数量是固定的,那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度,还节省了开闭进程的时间,也一定程度上能够实现并发效果。

如何创建?

Pool([numprocess  [,initializer [, initargs]]]):创建进程池

参数介绍:

1 numprocess: 要创建的进程数,如果省略,将默认使用cpu_count()的值
2 initializer:是每个工作进程启动时要执行的可调用对象,默认为None
3 initargs:是要传给initializer的参数组

主要方法

p.apply一般用于同步。

p.apply_asyn 一般用于异步

# 方法介绍
 # 1 p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
 # 2 ‘‘‘需要强调的是:此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数,
   必须从不同线程调用p.apply()函数或者使用p.apply_async()‘‘‘
 # 3
 # 4 p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
 # 5 ‘‘‘此方法的结果是AsyncResult类的实例,callback是可调用对象,接收输入参数。当func的结果变为可用时,
将理解传递给callback。callback禁止执行任何阻塞操作,否则将接收其他异步操作中的结果。‘‘‘
 # 6
 # 7 p.close():关闭进程池,防止进一步操作。如果所有操作持续挂起,它们将在工作进程终止前完成
 # 8
 # 9 P.jion():等待所有工作进程退出。此方法只能在close()或teminate()之后调用

其他方法(了解)

其其他方法(了解)
    # 1
    # 方法apply_async()
    # 和map_async()的返回值是AsyncResul的实例obj。实例具有以下方法
    # 2
    # obj.get(): 返回结果,如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达,将引发一场。
如果远程操作中引发了异常,它将在调用此方法时再次被引发。
    # 3
    # obj.ready(): 如果调用完成,返回True
    # 4
    # obj.successful(): 如果调用完成且没有引发异常,返回True,如果在结果就绪之前调用此方法,引发异常
    # 5
    # obj.wait([timeout]): 等待结果变为可用。
    # 6
    # obj.terminate():立即终止所有工作进程,同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收,
将自动调用此函数

例子:

#进程池的同步调用

#进程池的同步调用
import os,time#引入系统模块和时间模块
from multiprocessing import Pool#引入进程池模块

def work(n):#定义一个函数
print(‘%s run‘ %os.getpid())#打印id
time.sleep(3)#睡3秒
return n**2#返回一个n平方

if __name__ == ‘__main__‘:#如果文件名等于当前文件名
p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
res_l=[]#创建一个列表
for i in range(10):#循环十个数
res=p.apply(work,args=(i,)) # 同步调用,直到本次任务执行完毕拿到res,等待任务work执行的过程中可能有阻塞
        res_l.append( )               也可能没有阻塞
                                    # 但不管该任务是否存在阻塞,同步调用都会在原地等着
print(res_l)#打印列表

打印结果:

9656 run
6536 run
1492 run
9656 run
6536 run
1492 run
9656 run
6536 run
1492 run
9656 run
[]

想要有返回值,就在列表添加res,就能打印[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

import os
import time
import random
from multiprocessing import Pool

def work(n):
    print(‘%s run‘ %os.getpid())
    time.sleep(random.random())
    return n**2

if __name__ == ‘__main__‘:
    p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
res_l=[]
    for i in range(10):
        res=p.apply_async(work,args=(i,)) # 异步运行,根据进程池中有的进程数,每次最多3个子进程在异步执行
                                          # 返回结果之后,将结果放入列表,归还进程,之后再执行新的任务
                                          # 需要注意的是,进程池中的三个进程不会同时开启或者同时结束
                                          # 而是执行完一个就释放一个进程,这个进程就去接收新的任务。
res_l.append(res)

    # 异步apply_async用法:如果使用异步提交的任务,主进程需要使用jion,等待进程池内任务都处理完,
       然后可以用get收集结果
    # 否则,主进程结束,进程池可能还没来得及执行,也就跟着一起结束了
p.close()
    p.join()
    for res in res_l:
        print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,
                         立刻获取结果,也根本无需get

注意:不管是同步还是异步都有返回值。

原文地址:https://www.cnblogs.com/12121wdw/p/8666206.html

时间: 2024-10-06 20:16:01

管道,进程间数据共享,进程池的相关文章

Perl进程间数据共享

本文介绍的Perl进程间数据共享内容主体来自于<Pro Perl>的第21章. IPC简介 通过fork创建多个子进程时,进程间的数据共享是个大问题,要么建立一个进程间通信的通道,要么找到一个两进程都引用的共享变量.本文将介绍Unix IPC的近亲System V IPC:message queues(消息队列).semaphores(信号量)和shared memory-segments(共享内存段).它们都是IPC结构,它们被非常广泛地应用于进程间通信.它们的帮助文档可参见: $ perl

通过SharedPreferences实现进程间数据共享的问题详解

??之前为了解决应用的内存压力,在同一个应用中使用了多进程,但在程序自测的过程中发现不同进程之间的SharedPreferences数据不能共享,但应用内很多数据都是通过SharedPreferences来保存的,如果改成其它多进程通信的方式改动比较大.通过查看源码发现,在API Level>=11即Android 3.0可以通过Context.MODE_MULTI_PROCESS属性来实现SharedPreferences多进程共享,具体使用方式如下: SharedPreferences sp

Python之路(第三十九篇)管道、进程间数据共享Manager

一.管道 概念 管道可用于具有亲缘关系进程间的通信,有名管道克服了管道没有名字的限制,因此,除具有管道所具有的功能外,它还允许无亲缘关系进程间的通信. 先画一幅图帮助大家理解下管道的基本原理 现有2个进程A和B,他们都在内存中开辟了空间,那么我们在内存中再开辟一个空间C,作用是连接这两个进程的.对于进程来说内存空间是可以共享的(任何一个进程都可以使用内存,内存当中的空间是用地址来标记的,我们通过查找某一个地址就能找到这个内存)A进程可以不断的向C空间输送东西,B进程可以不断的从C空间读取东西,这

113 python程序中的进程操作-进程间数据共享(multiProcess.Manger)

一.进程之间的数据共享 展望未来,基于消息传递的并发编程是大势所趋 即便是使用线程,推荐做法也是将程序设计为大量独立的线程集合,通过消息队列交换数据. 这样极大地减少了对使用锁定和其他同步手段的需求,还可以扩展到分布式系统中. 但进程间应该尽量避免通信,即便需要通信,也应该选择进程安全的工具来避免加锁带来的问题. 以后我们会尝试使用数据库来解决现在进程之间的数据共享问题. 1.1 Manager模块介绍 虽然进程间数据独立,但可以通过Manager实现数据共享,事实上Manager的功能远不止于

Python并发编程-进程间数据共享

Manager中进程数据不安全 通过加锁解决 from multiprocessing import Manager,Process,Lock def main(dic,lock): lock.acquire() dic['count'] -=1 print(dic) lock.release() if __name__ == '__main__': m = Manager() l = Lock() dic = m.dict({'count':100}) #主进程中数据提供到子进程去操作 p_l

进程间的数据共享、进程池的回调函数和线程初识、守护线程

一.进程的数据共享 进程间数据是独立的,可以借助于队列或管道实现通信,二者都是基于消息传递的 虽然进程间数据独立,但可以通过Manager实现数据共享. 把所有实现了数据共享的比较便捷的类都重新又封装了一遍,并且在原有的multiprocessing基础上增加了新的机制 list dict等 数据共享的机制 支持数据类型非常有限 list dict都不是数据安全的,你需要自己加锁来保证数据安全 Manager用法: Manager().dict() # 创建共享的字典 Manager().lis

【进程间通讯与进程池】 -- 2019-08-11 18:46:25

原文: http://106.13.73.98/__/4/ 目录 一.队列 二.管道 三.进程间数据共享 四.进程池 进程间通讯:IPC(Inter-Process Communication) 一.队列: 队列:先进先出(First In First Out)简称 FIFO 栈:先进后出(First In Last Out)简称 FILO 1. multiprocessing.Queue模块 用于创建共享的进程队列,Queue是多进程安全的队列,可以实现对进程之间的数据传递,队列底层是使用管道

【进程间通讯与进程池】 -- 2019-08-16 18:12:52

原文: http://blog.gqylpy.com/gqy/230 " 目录 一.队列 二.管道 三.进程间数据共享 四.进程池 进程间通讯:IPC(Inter-Process Communication) 一.队列: 队列:先进先出(First In First Out)简称 FIFO 栈:先进后出(First In Last Out)简称 FILO 1. multiprocessing.Queue模块 用于创建共享的进程队列,Queue是多进程安全的队列,可以实现对进程之间的数据传递,队列

【进程间通讯与进程池】 &#223572;

原文: http://blog.gqylpy.com/gqy/230 " 目录 一.队列 二.管道 三.进程间数据共享 四.进程池 进程间通讯:IPC(Inter-Process Communication) 一.队列: 队列:先进先出(First In First Out)简称 FIFO 栈:先进后出(First In Last Out)简称 FILO 1. multiprocessing.Queue模块 用于创建共享的进程队列,Queue是多进程安全的队列,可以实现对进程之间的数据传递,队列