python并发编程(管道,事件,信号量,进程池)

Posted 2021-02-02 吾有一剑

管道

Conn1,conn2 = Pipe()

Conn1.recv()

Conn1.send()

数据接收一次就没有了

from multiprocessing import Process,Pipe

def f1(conn):

    from_zhujincheng = conn.recv()
    print(‘子进程‘)
    print(‘来自主进程的消息:‘,from_zhujincheng)

if __name__ == ‘__main__‘:
    conn1,conn2 = Pipe()  #创建一个管道对象,全双工,返回管道的两端,但是一端发送的消息,只能另外一端接收,自己这一端是不能接收的
    p1 = Process(target=f1,args=(conn2,))
    p1.start()

    conn1.send(‘出来吧‘)
    print(‘主进程‘)

事件

E = Event()  #初识状态是false

E.wait()  当事件对象e的状态为false的时候,在wait的地方会阻塞程序,当对象状态为true的时候,直接在这个wait地方继续往下执行

E.set()  将事件对象的状态改为true,

E.is_set() 查看状态

E.clear()  将事件对象的状态改为false

from multiprocessing import Process,Event

e = Event()  #创建事件对象,这个对象的初识状态为False
print(‘e的状态是:‘,e.is_set())  #  False

print(‘进程运行到这里了‘)
e.set()  #将e的状态改为True
print(‘e的状态是:‘,e.is_set())  #  True

e.clear()  #将e的状态改为False

e.wait()  #e这个事件对象如果值为False,就在我加wait的地方等待

print(‘进程过了wait‘)

信号量

S = semphore(数字),内部维护了一个计数器,acquire-1,release+1,为0的时候,其他的进程都要在acquire之前等待

S.acquire()

需要锁住的代码

S.release()

import time,random
from multiprocessing import Process,Semaphore

def f1(i,s):
    s.acquire()
    print(‘%s男嘉宾到了‘%i)
    time.sleep(random.randint(1,3))
    s.release()

if __name__ == ‘__main__‘:
    s = Semaphore(4)  #计数器4,acquire一次减一,为0 ,其他人等待,release加1
    for i in range(10):
        p = Process(target=f1,args=(i,s))
        p.start()

进程池

进程的创建和销毁是很有消耗的,影响代码执行效率

在有进程池的代码中,主进程运行结束,进程池里面的任务全部停止,不会等待进程池里面的任务

pl = Pool(数字)   这个数字一般是电脑的cpu数

pl的方法:

　　Map:异步提交任务,并且传参需要可迭代类型的数据,自带close和join功能

import time
from multiprocessing import Process,Pool

#对比多进程和进程池的效率
def f1(n):
    for i in range(5):
        n = n + i

if __name__ == ‘__main__‘:

    #统计进程池执行100个任务的时间
    s_time = time.time()
    pool = Pool(4)  
    pool.map(f1,range(100))  
    e_time = time.time()
    dif_time = e_time - s_time

    #统计100个进程,来执行100个任务的执行时间
    p_s_t = time.time() #多进程起始时间
    p_list = []
    for i in range(100):
        p = Process(target=f1,args=(i,))
        p.start()
        p_list.append(p)
    [pp.join() for pp in p_list]
    p_e_t = time.time()
    p_dif_t = p_e_t - p_s_t
    print(‘进程池的时间:‘,dif_time)
    print(‘多进程的执行时间:‘,p_dif_t)
    # 结果:  进程池的时间: 0.40102291107177734    多进程的执行时间: 9.247529029846191     
    # 可以看出进程池运行效率远远大于创建多进程

　　Close : 锁住进程池,防止有其他的新的任务在提交给进程池

　　Join : 等待着进程池将自己里面的任务都执行完

　　Res = Apply(f1,args=(i,))  #同步执行任务,必须等任务执行结束才能给进程池提交下一个任务,可以直接拿到返回结果res

import time
from multiprocessing import Process,Pool

def f1(n):
    time.sleep(1)
    return n*n

if __name__ == ‘__main__‘:

    pool = Pool(4)
    for i in range(10):
        res = pool.apply(f1,args=(i,))
        print(res)

　　Res_obj = Apply_async(f1,args=(i,))  #异步提交任务,可以直接拿到结果对象,从结果对象里面拿结果,要用get方法,get方法会阻塞程序,没有拿到结果会一直等待

import time
from multiprocessing import Process,Pool

def f1(n):
    time.sleep(0.5)
    return n*n

if __name__ == ‘__main__‘:

    pool = Pool(4)

    res_list = []
    for i in range(10):
        res = pool.apply_async(f1,args=(i,)) # 不能直接打印返回值,因为直接返回结果对象,进程还没执行完,结果对象里没有数据
        res_list.append(res)

    pool.close()  
    pool.join()

    #打印结果,异步提交之后的结果对象
    for i in res_list:
        print(i.get())

 

回调函数:

 Apply_async(f1,args=(i,),callback=function)  #将前面f1这个任务的返回结果作为参数传给callback指定的那个function函数

import os
from multiprocessing import Pool,Process

def f1(n):
    print(‘传入的函数‘,n)
    return n*n

def call_back_func(asdf):
    print(‘回调函数‘,asdf)

if __name__ == ‘__main__‘:
    pool = Pool(4)
    res = pool.apply_async(f1,args=(5,),callback=call_back_func)
    pool.close()
    pool.join()