multiprocess模块补充

Posted dymlnet

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了multiprocess模块补充相关的知识,希望对你有一定的参考价值。

进程间通信——队列和管道(multiprocess.Queue、multiprocess.Pipe)

在电脑的多个进程中,他们之间要想通讯需要使用IPC协议(Inter-Process Communication)

Queue([maxsize]) 
创建共享的进程队列。
参数 :maxsize是队列中允许的最大项数。如果省略此参数,则无大小限制。
底层队列使用管道和锁定实现。
Queue([maxsize]) 
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数,则无大小限制。底层队列使用管道和锁定实现。另外,还需要运行支持线程以便队列中的数据传输到底层管道中。 
Queue的实例q具有以下方法:

q.get( [ block [ ,timeout ] ] ) 
返回q中的一个项目。如果q为空,此方法将阻塞,直到队列中有项目可用为止。block用于控制阻塞行为,默认为True. 如果设置为False,将引发Queue.Empty异常(定义在Queue模块中)。
timeout是可选超时时间,用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用,将引发Queue.Empty异常。 q.get_nowait( ) 同q.get(False)方法。 q.put(item [, block [,timeout ] ] ) 将item放入队列。如果队列已满,此方法将阻塞至有空间可用为止。block控制阻塞行为,默认为True。如果设置为False,将引发Queue.Empty异常(定义在Queue库模块中)。
timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。 q.qsize() 返回队列中目前项目的正确数量。此函数的结果并不可靠,因为在返回结果和在稍后程序中使用结果之间,队列中可能添加或删除了项目。在某些系统上,此方法可能引发NotImplementedError异常。 q.empty() 如果调用此方法时 q为空,返回True。如果其他进程或线程正在往队列中添加项目,结果是不可靠的。也就是说,在返回和使用结果之间,队列中可能已经加入新的项目。 q.full() 如果q已满,返回为True. 由于线程的存在,结果也可能是不可靠的(参考q.empty()方法)。。
q.close() 
关闭队列,防止队列中加入更多数据。调用此方法时,后台线程将继续写入那些已入队列但尚未写入的数据,但将在此方法完成时马上关闭。

如果q被垃圾收集,将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如,如果某个使用者正被阻塞在get()操作上,
关闭生产者中的队列不会导致get()方法返回错误。 q.cancel_join_thread() 不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。 q.join_thread() 连接队列的后台线程。此方法用于在调用q.close()方法后,等待所有队列项被消耗。默认情况下,此方法由不是q的原始创建者的所有进程调用。
调用q.cancel_join_thread()方法可以禁止这种行为。

 

生产者消费者模型

非常典型的案例

# 矛盾 : 解决 数据的供需不平衡
# 同步 生产数据 使用数据
# 异步 主进程 生产数据 子进程使用数据 —— 500个子进程去处理数据 生产过慢
# 异步 子进程 生产数据 子进程处理数据

# 3个子进程 生产泔水
# 2个子进程 吃泔水
# 生产的快 吃的慢 泔水溢出
# 生产的慢 吃得快 泔水不够吃
# 如果增加生产者 —— 让生产变快
# 减少生产者 —— 找一个容器放 约束容器的容量

import time
import random
from multiprocessing import Queue
from multiprocessing import Process
def producer(q,food):
    for i in range(5):
        q.put(%s-%s%(food,i))
        print(生产了%s%food)
        # time.sleep(random.randint(1,3))
        time.sleep(random.random())
    q.put(None)
    q.put(None)
    q.put(None)

def consumer(q,name):
    while True:
        food = q.get()   # 生产者不生产还是生产的慢
        if food == None:break
        print(%s 吃了 %s%(name,food))

if __name__ == __main__:
    q = Queue()
    p1 = Process(target=producer,args=(q,泔水))
    p1.start()
    p2 = Process(target=producer, args=(q, 骨头))
    p2.start()
    c1 = Process(target=consumer, args=(q, A))
    c1.start()
    c2 = Process(target=consumer, args=(q, B))
    c2.start()
    c3 = Process(target=consumer, args=(q, C))
    c3.start()

# 队列不会不安全
# 现在 通过queue
# 生产者消费者模型
#1.消费者要处理多少数据是不确定的
#2.所以只能用while循环来处理数据 ,但是while循环无法结束
#3.需要生产者发送信号
#4.有多少个消费者 就需要发送多少个信号
#5.但是发送的信号数量需要根据 生产者和消费者的数量进行计算,所以非常不方便

 

完善版

import time
import random
from multiprocessing import Process
from multiprocessing import JoinableQueue
# put   q.join
# get  处理数据 task_done 消费完了
def producer(q,food):
    for i in range(5):
        q.put(%s-%s%(food,i))
        print(生产了%s%food)
        # time.sleep(random.randint(1,3))
        time.sleep(random.random())
    q.join()  # 等待 消费者 把所有的数据都处理完

def consumer(q,name):
    while True:
        food = q.get()   # 生产者不生产还是生产的慢
        print(%s 吃了 %s%(name,food))
        q.task_done()

if __name__ == __main__:
    q = JoinableQueue()
    p1 = Process(target=producer,args=(q,泔水))
    p1.start()
    p2 = Process(target=producer, args=(q, 骨头))
    p2.start()
    c1 = Process(target=consumer, args=(q, alex))
    c1.daemon = True
    c1.start()
    c2 = Process(target=consumer, args=(q, egon))
    c2.daemon = True
    c2.start()
    c3 = Process(target=consumer, args=(q, jin))
    c3.daemon = True
    c3.start()

    p1.join()  # 等待p1执行完毕
    p2.join()  # 等待p2执行完毕

# 生产者生产的数据全部被消费 —— 生产者进程结束 —— 主进程代码执行结束 —— 消费者守护进程结束

JoinableQueue  模块

在队列基础上增加了返回,task_done()返回执行完毕的信号

 

管道,不加锁的队列、会导致数据紊乱和不安全

manager  计数

from multiprocessing import Manager,Process,Lock
def work(d,lock):
    lock.acquire()
    d[count] -= 1
    lock.release()

if __name__ == __main__:
        lock= Lock()
        m = Manager()
        dic=m.dict({count:100})   # 共享的数据
        l = []
        for i in range(100):
            p=Process(target=work,args=(dic,lock))
            p.start()
            l.append(p)
        [p.join() for p in l]
        print(dic)

 

利用Manager模块来达到所有进程共用亿个dic中的count值,我们可以看到最后输出的dic是0

技术分享图片

但是他本身并不安全

 






















以上是关于multiprocess模块补充的主要内容,如果未能解决你的问题,请参考以下文章

13多进程multiprocessing进程池

multiprocessing_generator 模块触发权限错误

VSCode 配置 用户自定义代码片段 自定义自动代码补充

multiprocessing模块

Python多进程 - subprocess & multiprocess

multiprocessing模块