#一、进程Process#在windows中使用需要注意#在Windows操作系统中由于没有fork(linux操作系统中创建进程的机制)，在创建子进程的时候会自动 import 启动它的这个文件，而在 import 的时候又执行了整个文件。因此如果将process()直接写在文件中就会无限递归创建子进程报错。#所以必须把创建子进程的部分使用if __name__ =='__main__' 判断保护起来，import 的时候 ，就不会递归运行了。#join：父进程等待子进程结束后才继续执行自己后续的代码

importtimeimportrandomfrom multiprocessing importProcessdeffunc(index):

time.sleep(random.randint(1, 3))print('第%s封邮件发送完毕' %index)if __name__ == '__main__':

p_lst=[]for i in range(10):

p= Process(target=func, args=(i,))

p.start()#先让所有子进程都启动

p_lst.append(p)for p in p_lst: #再进行join阻塞

p.join()print('10封邮件全部发送完毕')#守护进程#主进程创建守护进程#1：守护进程会在主进程代码执行结束后就终止#2：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children#注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止

#1，守护进程会在主进程代码执行结束后就终止

importtimefrom multiprocessing importProcessdeffunc():print('子进程 start')

time.sleep(3) #睡3秒的时候主进程的代码已经执行完毕了，所以子进程也会跟着结束

print('子进程end')if __name__ == '__main__':

p= Process(target=func)

p.daemon= True #daemon是Process的属性

p.start()

time.sleep(2) #睡2秒的时候，执行了子进程

print('主进程')#结果：#子进程 start#主进程

#锁#加锁降低了程序的效率，让原来能够同时执行的代码变成顺序执行了，异步变同步的过程#保证了数据的安全

importtimeimportjsonfrom multiprocessing importProcessfrom multiprocessing import Lock #导入Lock类

defsearch(person):

with open('ticket') as f:

ticketinfo=json.load(f)print('%s查询余票：' % person, ticketinfo['count'])defget_ticket(person):

with open('ticket') as f:

ticketinfo=json.load(f)

time.sleep(0.2) #模拟读数据的网络延迟

if ticketinfo['count'] >0:print('%s买到票了' %person)

ticketinfo['count'] -= 1time.sleep(0.2)

with open('ticket', 'w') as f:

json.dump(ticketinfo, f)else:print('%s没买到票' %person)defticket(person, lock):

search(person)

lock.acquire()#谁获得钥匙 谁才能进入

get_ticket(person)

lock.release()#用完了，把钥匙给下一个人

if __name__ == '__main__':

lock= Lock() #创建一个锁对象

for i in range(5):

p= Process(target=ticket, args=('person%s' %i, lock))

p.start()#结果：#person1查询余票： 3#person3查询余票： 3#person0查询余票： 3#person2查询余票： 3#person4查询余票： 3#person1买到票了#person3买到票了#person0买到票了#person2没买到票#person4没买到票

#1、信号量的实现机制：计数器 + 锁实现的#信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()#调用被阻塞。#互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据(Samphore相当于有几把钥匙，lock只能有一把钥匙)

importtimeimportrandomfrom multiprocessing importProcessfrom multiprocessing importSemaphoredef changba(person, sem): #在唱吧 唱歌

sem.acquire() #第一次可以同时进来两个人

print('%s走进唱吧' %person)

time.sleep(random.randint(3, 6)) #每个人唱歌的时间

print('%s走出唱吧' % person) #唱完走人

sem.release() #把钥匙给下一个人

if __name__ == '__main__':

sem= Semaphore(2) #2把钥匙

for i in range(5):

p= Process(target=changba, args=('person%s' %i, sem))

p.start()#事件 Event#事件主要提供了三个方法#set、wait、clear。#事件处理的机制：全局定义了一个“Flag”，#如果“Flag”值为False，那么当程序执行event.wait方法时就会阻塞，#如果“Flag”值为True，那么event.wait方法时便不再阻塞。

#阻塞事件 ：wait()方法#wait是否阻塞是看event对象内部的Flag

#控制Flag的值:#set() 将Flag的值改成True#clear() 将Flag的值改成False#is_set() 判断当前的Flag的值

#红绿灯：

importtimeimportrandomfrom multiprocessing importProcessfrom multiprocessing importEventdef traffic_ligth(e): #红绿灯

print('\033[31m红灯亮\033[0m') #Flag 默认是False

whileTrue:if e.is_set(): #如果是绿灯

time.sleep(2) #2秒后

print('\033[31m红灯亮\033[0m') #转为红灯

e.clear() #设置为False

else: #如果是红灯

time.sleep(2) #2秒后

print('\033[32m绿灯亮\033[0m') #转为绿灯

e.set() #设置为True

def car(e, i): #车

if note.is_set():print('car %s在等待' %i)

e.wait()print('car %s 通过了' %i)if __name__ == '__main__':

e=Event()

p= Process(target=traffic_ligth, args=(e,)) #红绿灯进程

p.daemon =True

p.start()

p_lst=[]for i in range(10): #10辆车的进程

time.sleep(random.randrange(0, 3, 2))

p= Process(target=car, args=(e, i))

p.start()

p_lst.append(p)for p inp_lst: p.join()#二、进程通信#队列(先进先出)#生产者消费者模型

importtimeimportrandomfrom multiprocessing importProcess, Queuedefconsumer(q, name):#消费者

whileTrue:

food= q.get() #在队列中取值

if food is None: breaktime.sleep(random.uniform(0.3, 1)) #模拟吃消耗的时间

print('%s偷吃了%s，快打死他' %(name, food))defproducter(q, name, food):#生产者

for i in range(10):

time.sleep(random.uniform(0.5, 0.9)) #模拟生产时间

print('%s生产了%s，序号：%s' %(name, food, i))

q.put(food+ str(i)) #把值存入队列中

if __name__ == '__main__':

q= Queue() #Queue队列对象

c1 = Process(target=consumer, args=(q, '小明'))

c2= Process(target=consumer, args=(q, '小东'))

c1.start()

c2.start()

p1= Process(target=producter, args=(q, '张三', '面包'))

p2= Process(target=producter, args=(q, '李四', '可乐'))

p1.start()

p2.start()

p1.join()

p2.join()

q.put(None)#有几个consumer进程就需要放几个None,表示生产完毕(这就有点low了)

q.put(None)#JoinableQueue#JoinableQueue和Queue几乎一样，不同的是JoinableQueue队列允许使用者告诉队列某个数据已经处理了。通知进程是使用共享的信号和条件变量来实现的。#task_done()：使用者使用此方法发出信号，表示q.get()返回的项目已经被处理#join()：当队列中有数据的时候，使用此方法会进入阻塞，直到放入队列中所有的数据都被处理掉(都被task_done)才转换成不阻塞

#解决刚才生产者消费者模型low的问题：

importtimeimportrandomfrom multiprocessing importProcess, JoinableQueuedefconsumer(jq, name):#消费者

whileTrue:

food= jq.get() #在队列中取值

#if food is None:break

time.sleep(random.uniform(0.3, 1)) #模拟吃消耗的时间

print('%s偷吃了%s，快打死他' %(name, food))

jq.task_done()#向jq.join()发送一次信号,证明这个数据已经处理了

defproducter(jq, name, food):#生产者

for i in range(10):

time.sleep(random.uniform(0.5, 0.9)) #模拟生产时间

print('%s生产了%s，序号：%s' %(name, food, i))

jq.put(food+ str(i)) #把值存入队列中

if __name__ == '__main__':

jq=JoinableQueue()

c1= Process(target=consumer, args=(jq, '小明'))

c2= Process(target=consumer, args=(jq, '小东'))

c1.daemon= True #把消费者设置为守护进程

c2.daemon =True

c1.start()

c2.start()

p1= Process(target=producter, args=(jq, '张三', '面包'))

p2= Process(target=producter, args=(jq, '李四', '可乐'))

p1.start()

p2.start()

p1.join()

p2.join()

jq.join()#数据全部被task_done后才不阻塞

#管道#Pipe([duplex]):在进程之间创建一条管道，并返回元组(left,right),其中left,right表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道#duplex:默认管道是全双工的，如果将duplex改成False，left只能用于接收，right只能用于发送。#主要方法：#right.recv(): 接收left.send()发送的内容。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError。#letf.send(): 通过连接发送内容。#close(): 关闭连接。

#pipe的端口管理不会随着某一个进程的关闭就关闭#操作系统来管理进程对这些端口的使用，不使用的端口应该关闭它#一条管道，两个进程，就有4个端口 每关闭一个端口计数-1，直到只剩下一个端口的时候 recv就会报错#如果不关闭不使用的端口，在已经把数据发送完毕的情况下，那么接收端的recv就会一直挂起，等待接收数据，这个进程就一直不能关闭#因此不使用的端口就应该关闭它，让recv抛出异常后对这个进程进行处理

from multiprocessing importProcess, Pipedefconsumer(left, right):

left.close()#若这里不close，则不会异常EOFError，数据接收完毕后，下面的right.recv()就会一直挂起

whileTrue:try:print(right.recv())exceptEOFError:break

if __name__ == '__main__':

left, right=Pipe()

Process(target=consumer, args=(left, right)).start()

right.close()for i in range(10):

left.send('Apple%s' %i)

left.close()#三、进程池#同步、apply

importosimporttimefrom multiprocessing importPooldeftest(num):

time.sleep(1)print('%s:%s' %(num, os.getpid()))return num * 2

if __name__ == '__main__':

p=Pool()for i in range(20):

res= p.apply(test, args=(i,)) #提交任务的方法 同步提交

print('-->', res) #res就是test的return的值

#异步、apply_async

importtimefrom multiprocessing importPooldeffunc(num):

time.sleep(1)print('做了%s件衣服' %num)if __name__ == '__main__':

p= Pool(4) #进程池中创建4个进程，不写的话，默认值为你电脑的CUP数量

for i in range(50):

p.apply_async(func, args=(i,)) #异步提交func到一个子进程中执行，没有返回值的情况

p.close() #关闭进程池，用户不能再向这个池中提交任务了

p.join() #阻塞，直到进程池中所有的任务都被执行完

#注意：#异步提交且没有返回值接收的情况下必须要用close()和join()#因为如果没有close()和join()，主进程执行完毕后会立刻把子进程回收了，相当于子进程还没来得及开启#所以要join，让子进程结束后再结束父进程，但是进程池中要使用join就必须先进行close

importtimeimportosfrom multiprocessing importPooldeftest(num):

time.sleep(1)print('%s:%s' %(num, os.getpid()))return num * 2

if __name__ == '__main__':

p=Pool()

res_lst=[]for i in range(20):

res= p.apply_async(test, args=(i,)) #提交任务的方法 异步提交

res_lst.append(res)for res inres_lst:print(res.get())#注意：#异步提交有返回值的情况下，res是一个对象代表的是这个任务的编号，需要用res.get()方法让任务执行且把返回值返回给你。#get有阻塞效果，拿到子进程的返回值后才不阻塞，所以并不需要再使用close和join。

#map#map接收一个函数和一个可迭代对象，是异步提交的简化版本，自带close和join方法#可迭代对象的每一个值就是函数接收的实参，可迭代对象的长度就是创建的任务数量#map拿到返回值是所有结果组成的列表

importtimefrom multiprocessing importPooldeffunc(num):print('子进程：', num)#time.sleep(1)

returnnumif __name__ == '__main__':

p=Pool()

ret= p.map(func, range(10)) #ret是列表

for i inret:print('返回值：', i)#进程池的回调函数(同步提交apply没有回调函数)

importosfrom multiprocessing importPooldeffunc(i):print('子进程：', os.getpid())returnidefcall_back(res):print('回调函数：', os.getpid())print('res--->', res)if __name__ == '__main__':

p=Pool()print('主进程：', os.getpid())

p.apply_async(func, args=(1,), callback=call_back) #callback关键字传参，参数是回调函数

p.close()

p.join()#结果：#主进程： 4732#子进程： 10552#回调函数： 4732#res---> 1#

#从结果可以看出：#子进程func执行完毕之后才去执行callback回调函数#子进程func的返回值会作为回调函数的参数#回调函数是在主进程中执行的