python-45-管道与进程池的使用

# 1、初识管道,可以互相通信。
from multiprocessing import Pipe
conn1, conn2 = Pipe()
conn1.send(‘123‘)
print(conn2.recv())
conn2.send(‘321‘)
print(conn1.recv())

# 2、管道实现：生产者/消费者模型
from multiprocessing import Pipe,Process
import time,random
def consumer(con,pro,name):
    pro.close()
    while 1:
        try:
            m=con.recv()
            print(‘%s使用了 %s‘ % (name,m))
            time.sleep(random.random())
        except EOFError:
            con.close()
            break
def producer(con,pro,name,mask):
    con.close()
    for i in range(1,6):
        time.sleep(random.random())
        m=‘%s生产了%s %s‘%(name,mask,i)
        print(m)
        pro.send(m)
    pro.close()
if __name__ == ‘__main__‘:
    con,pro=Pipe()
    p=Process(target=producer,args=(con,pro,‘大厂‘,‘N95 ‘))
    p.start()
    c=Process(target=consumer,args=(con,pro,‘A企业‘))
    c.start()
    c1=Process(target=consumer,args=(con,pro,‘B企业‘))
    c1.start()
    con.close()
    pro.close()

# 3、Manager,数据共享不安全的，但会进程之间抢占资源。
# 但可以加锁约束解决
from multiprocessing import Manager,Process,Lock
def func(dic,lock):
    # lock.acquire()
    dic[‘count‘]+=1
    # lock.release()

if __name__ == ‘__main__‘:
    lock=Lock()
    m=Manager()
    dic=m.dict({‘count‘:0})
    p_lst=[]
    for i in range(50):
        p=Process(target=func,args=(dic,lock))
        p.start()
        p_lst.append(p)
    for i in p_lst:i.join()
    print(‘主进程：%s‘%dic)

from multiprocessing import Pool,Process
import time
def func(i):
    print(i)

if __name__ == ‘__main__‘:
    # 进程池的效果
    st=time.time()
    pool=Pool(5)                # 池子可放5个(一般CPU的个数+1)
    pool.map(func,range(100))   # 100个进程任务
    t1=time.time()-st
    st=time.time()
    print(‘进程池时间：‘,t1)

    # 原来多进程的效果
    p_lst=[]
    for i in range(100):
        p=Process(target=func,args=(i,))
        p_lst.append(p)
        p.start()
    for p in p_lst:p.join()
    t2=time.time()-st
    print(‘多进程时间：‘,t2)

#  2、进程池传多个参数：
from multiprocessing import Pool
import time
def func(i):
    print(i)

if __name__ == ‘__main__‘:
    st=time.time()
    pool=Pool(5)
    pool.map(func,[(10,‘name‘,‘age‘),100])

# 3、apply：同步
from multiprocessing import Pool
import time
def func():
    print(‘--开始~‘)
    time.sleep(0.1)
    print(‘==结束！‘+‘\n‘)
if __name__ == ‘__main__‘:
    pool=Pool()
    for i in range(5):
        pool.apply(func)       # 同步了

# 4、apply_async：异步
from multiprocessing import Pool
import time,os
def func(i):
    pid=os.getpid()
    print(‘%s--开始~:%s‘%(i,pid))
    time.sleep(1)
if __name__ == ‘__main__‘:
    pool=Pool(2)
    for i in range(5):
        pool.apply_async(func,(i,))
    pool.close()        # 结束进程池接收任务
    pool.join()         # 感知进程池中的任务执行结束

# 5、进程池的返回值
# apply:直接接收返回值
# apply_async:需要get(),会堵塞因等待返回值，解决可先放列表get()。
from multiprocessing import Pool
import time
def func(i):
    time.sleep(0.5)
    return i+1
if __name__ == ‘__main__‘:
    p=Pool(3)
    p_lst=[]

    # for i in range(10):
        # res=p.apply(func,args=(i,))        # 直接接收返回值
        # print(res)

    #     res=p.apply_async(func,args=(i,))   # apply_async
    #     p_lst.append(res)
    # for i in p_lst:print(i.get())

    res=p.map(func,range(10))               # map一次性返回 
    print(res)

# 6、进程池回调函数
# 先执行异步函数func，将func返回值传入func1函数中的ii参数。
# 回调函数是在主进程中执行，而不是子进程中。
from multiprocessing import Pool
def func(i):
    return i
def func1(ii):
    print(‘i+1=‘,ii+1)
if __name__ == ‘__main__‘:
    p=Pool()
    for i in range(5):
        p.apply_async(func,args=(i,),callback=func1)
    p.close()
    p.join()