一、Process
参数介绍：
1 group参数未使用，值始终为None
2 target表示调用对象，即子进程要执行的任务
3 args表示调用对象的位置参数元组，args=(1,2,'a',)
4 kwargs表示调用对象的字典,kwargs={'name':'a','age':18}
5 name为子进程的名称
方法介绍：
1 p.start()：启动进程，并调用该子进程中的p.run()
2 p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
3 p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
4 p.is_alive():如果p仍然运行，返回True
5 p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
属性介绍：
1 p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
2 p.name:进程的名称
3 p.pid：进程的pid
4 p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)
5 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）
1、join：父进程等待子进程结束后才开始执行自己的代码
# 发送一封邮件
import time
import random
from multiprocessing import Process
def func():
time.sleep(random.randint(1,3)) # 模拟邮件发送的延迟
print('邮件已发送')
if __name__ == '__main__':
p = Process(target=func)
p.start()
p.join() # 阻塞 直到进程p执行完毕后才结束阻塞
print('邮件发送完毕')
# 发送十封邮件
import time
import random
from multiprocessing import Process
def func(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封邮件全部发送完毕')
2、用类的方式开启进程
我们之前创建进程的时候，其实也是在创建一个Process类的对象，再调用对象的start方法开启进程，
那么我们也可以自己定义一个类来实现进程的创建：
import os
from multiprocessing import Process
class MyProcess(Process): # 定义一个类，继承Process类
def run(self): # 必须实现的方法，是启动进程的方法
print('子进程:',os.getpid(),os.getppid())
if __name__ == '__main__':
p = MyProcess() # 实例化
p.start() # 自动调用run方法
print('父进程:',os.getpid())
给自定义类传参：
首先看看Process源码
import time
import os
from multiprocessing import Process
class MyProcess(Process):
def __init__(self,i):
super().__init__() # 实现父类的初始化方法
self.index = i # 定义自己的属性(参数)
def run(self):
time.sleep(1)
print('子进程:',self.index,os.getpid(),os.getppid())
if __name__ == '__main__':
p_lst = []
for i in range(10):
p = MyProcess(i)
p.start()
p_lst.append(p)
for p in p_lst:
p.join()
print('主进程:',os.getpid())
3、守护进程
主进程创建守护进程
1：守护进程会在主进程代码执行结束后就终止
2：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止
1，守护进程会在主进程代码执行结束后就终止
import time
from multiprocessing import Process
def func():
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
主进程
2，守护进程会在主进程代码执行结束后就终止 ，但是父进程会等待子进程结束才正式结束。
注意：代码结束是指代码运行到了最后一行，并不代表进程已经结束了。
import time
from multiprocessing import Process
def func():
count = 1
while True:
print('*' * count)
time.sleep(1)
count += 1
def func2():
print('普通进程开始')
time.sleep(5)
print('普通进程结束')
if __name__ == '__main__':
p1 = Process(target=func)
p1.daemon = True
p1.start()
Process(target=func2).start()
time.sleep(3)
print('主进程')
结果：
*
普通进程开始
**
***
主进程
普通进程结束
3，守护进程的作用
守护进程可以报活，就是向某个服务报告自己还活着
场景：
例如你写好了一个网页，你的服务端是不应该停的，因为你服务端一旦停止了，别人就无法访问你的网页了，所以我们应该确保服务端没有‘死’，
这个时候就可以使用守护进程，在你的服务端起一个守护进程，让这个进程只做一件事，就是每隔1个小时(时间按照自己的合理安排设定)向某一台机器汇报自己还活着，
一旦这个机器没有收到你守护进程传来的消息，那么就可以认为你的服务端已经挂了。
例如：
import time
from multiprocessing import Process
def Guard():
while True:
time.sleep(3600)
print('我还活着') # 向某个机器汇报我还活着，具体该怎么写汇报的逻辑就怎么写，这里只是示范
if __name__ == '__main__':
p = Process(target=Guard)
p.daemon = True
p.start()
# 主进程的逻辑(主进程应该是一直运行的，不应该有代码结束的时候)
print('主进程')
4、terminate：关闭进程
import time
from multiprocessing import Process
def fun():
print('子进程')
if __name__ == '__main__':
p = Process(target=fun)
p.start()
p.terminate() # 关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活
print(p.is_alive()) # True
time.sleep(0.1)
print(p.is_alive()) # False
二、锁 Lock
1、异步的问题
我们都知道异步进程的好处就是可以一起执行，效率高，但是当多个进程使用同一份数据资源的时候，就会引发数据安全或顺序混乱问题。
抢票系统：
#文件ticket的内容为：{"count":3}
#注意一定要用双引号，不然json无法识别
#并发运行，效率高，但竞争写同一文件，数据写入错乱
import time
import json
from multiprocessing import Process
def search(person):
with open('ticket') as f:
ticketinfo = json.load(f)
print('%s查询余票：' %person,ticketinfo['count'])
def get_ticket(person):
with open('ticket') as f:
ticketinfo = json.load(f)
time.sleep(0.2) #模拟读数据的网络延迟
if ticketinfo['count'] > 0:
print('%s买到票了'%person)
ticketinfo['count'] -= 1
time.sleep(0.2)
with open('ticket','w') as f:
json.dump(ticketinfo,f)
else:
print('%s没买到票'%person)
def ticket(person):
search(person)
get_ticket(person)
if __name__ == '__main__':
for i in range(5):
p = Process(target=ticket,args=('person%s'%i,))
p.start()
结果：
person0查询余票： 3
person4查询余票： 3
person1查询余票： 3
person2查询余票： 3
person3查询余票： 3
person0买到票了
person4买到票了
person1买到票了
person2买到票了
person3买到票了
分析：票只有三张，但是5个人都显示买到了，这是因为5个进程异步进行，大家都同一时间在对一个文件进行修改，导致的混乱。
2、用锁解决：
# 加锁降低了程序的效率，让原来能够同时执行的代码变成顺序执行了，异步变同步的过程
# 保证了数据的安全
import time
import json
from multiprocessing import Process
from multiprocessing import Lock # 导入Lock类
def search(person):
with open('ticket') as f:
ticketinfo = json.load(f)
print('%s查询余票：' %person,ticketinfo['count'])
def get_ticket(person):
with open('ticket') as f:
ticketinfo = json.load(f)
time.sleep(0.2) #模拟读数据的网络延迟
if ticketinfo['count'] > 0:
print('%s买到票了'%person)
ticketinfo['count'] -= 1
time.sleep(0.2)
with open('ticket','w') as f:
json.dump(ticketinfo,f)
else:
print('%s没买到票'%person)
def ticket(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没买到票
三、信号量 Semaphore
1、信号量的实现机制：计数器 + 锁实现的
信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。
互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据(Samphore相当于有几把钥匙，lock只能有一把钥匙)
import time
import random
from multiprocessing import Process
from multiprocessing import Semaphore
def 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
python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 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的值
红绿灯：
import time
import random
from multiprocessing import Process
from multiprocessing import Event
def traffic_ligth(e): # 红绿灯
print(' 33[31m红灯亮 33[0m') # Flag 默认是False
while True:
if e.is_set(): # 如果是绿灯
time.sleep(2) # 2秒后
print(' 33[31m红灯亮 33[0m') # 转为红灯
e.clear() # 设置为False
else: # 如果是红灯
time.sleep(2) # 2秒后
print(' 33[32m绿灯亮 33[0m') # 转为绿灯
e.set() # 设置为True
def car(e,i): # 车
if not e.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 in p_lst:p.join()
五、总结
进程之间虽然内存不共享，但是是可以通信的
Lock Semaphore Event 都在进行进城之间的通信
只不过这些通信的内容我们不能改变
后续还有队列和管道能让进程之间进行通信
