并发编程(一) 操作系统基础和进程 1.操作系统基础知识 2.进程基本概念 3.创建进程的两种方式 4.进程方法join 5.进程间数据隔离问题 6.进程对象及其他方法 7.僵尸进程和孤儿进程 8.守护进程 8.互斥锁 9.进程间通信
一.操作系统的作用
1.隐藏丑陋复杂的硬件接口,提供良好的抽象接口
2.管理、调度进程,并且将多个进程对硬件的竞争变得有序
二.多道技术
1.空间上的复用
多个程序共用一套计算机硬件
2.时间上的复用
切换+保存状态
1.当一个程序遇到I/O操作时,操作系统会剥夺该程序的cpu执行权限(提高cpu的利用率,也不会影响程序执行效率)
2.当一个程序长时间占用cpu,操作系统会剥夺该程序的cpu执行权限(降低了程序的执行效率)
2.进程基本概念
什么是进程
程序就是一串代码,进程就是正在运行的程序,它是资源分配和调度的基本单位,是操作系统结构的基础.
进程调度的算法
1.先来先服务调度算法
2.短作业优先调度算法
3.时间片轮转法
4.多级反馈队列
第一种和第二种算法都有明显的缺点,第一种对短作业不友好,第二种对长作业不友好,所以现代计算机进程都是基于第三种和第四种方法实现进程,如下图所示
进程的并行和并发
并发:看起来像同时运行
并行:真正意义上的同时执行
注意:单核计算机不能实现并行,但可以实现并发,所以要实现并行必须有多个处理器
进程的三种状态
1.就绪态:
当进程已经做好了所有准备,就等cpu执行了,此时的状态就是就绪态
2.运行态
当进程正在被cpu执行的时候,此时的状态就是执行态
3.阻塞态:
当进程在被执行的过程中发生了一些必要的I/O操作无法继续执行时,cpu会放弃该进程转而执行其他进程,此时的状态就是阻塞态
1 # 此时程序处于就绪态 2 import time # 运行这串代码时处于运行态 3 4 print('程序开始执行') 5 msg = input('>>>:') # 进入阻塞态 6 # 进入就绪态 7 print(msg) # 进入运行态 8 time.sleep(1) # 进入阻塞态 9 # 进入就绪态 10 print('程序结束运行') # 进入运行态 11 # 结束运行
同步异步
同步就是一个任务的完成必须等待另一个任务完成后才能算完成(类似于排队,程序的层面就是卡住了)
异步就是一个任务在执行的完成不需要知道另一个任务是否完成,只要自己的任务完成就算完成了(类似于叫号,另一个任务的完成结果是要的,但是是通过其他方式获取)
阻塞非阻塞
阻塞就是进程不在执行,处于阻塞态
非阻塞就是进程还在执行,处于就绪态或者运行态
两者组合成为四种状态
1.同步阻塞:只能做一件事情,并且处于阻塞态,效率最低
2.异步阻塞:因为等待某个事件而处于阻塞态,但是他不依赖于其他任务,所以事件处理完成就可以了
3.同步非阻塞:需要依赖于另一个任务的完成,就算自己的任务先完成了,但是还得等其他任务,效率也是地下的
4.异步非阻塞:不需要依赖于其他任务,自己又是非阻塞状态,可想而知效率是最高的
3.创建进程的两种方式
在python中创建进程需要导入一个multiprocess模块,他是python中操作、管理进程的一个包,他包含了和进程有关的所有子模块
multiprocess.process模块
process模块是一个创建进程的模块
创建进程就是在内存中重新开辟一块属于进程的独立的内存空间
进程和进程之间的数据是隔离的,无法直接交互,但是可以通过某些技术实现间接交互
1 from multiprocessing import Process 2 import time 3 4 age = 18 5 6 def run(msg): 7 global age 8 age = 28 9 print('%s开始'%msg) 10 print(age) # 子进程中的是28,两者无法访问 11 time.sleep(1) 12 print('%s结束'%msg) 13 14 ''' 15 windows会在进程创建时会以模块的方式从上到下的执行一遍, 16 所以在windows中创建进程一定要在if __name__ == '__main__':代码块中创建 17 linux中创建进程会直接复制一份 18 ''' 19 if __name__ == '__main__': 20 p = Process(target=run,args=('子进程',)) # 生成一个进程对象 21 p.start() # 创建一个进程 22 print(age) # 主进程中的age是18 23 print('主进程') # 主进程中的内容
1 from multiprocessing import Process 2 import time 3 age = 18 4 # 新建一个类继承父类Process 5 class MyProcess(Process): 6 def __init__(self,msg): 7 super().__init__() 8 self.msg = msg 9 10 def run(self): 11 global age 12 age = 28 13 print('%s开始'%self.msg) 14 print(age) 15 time.sleep(1) 16 print('%s结束'%self.msg) 17 18 if __name__ == '__main__': 19 p = MyProcess('子进程',) 20 p.start() 21 print(age) 22 print('主进程')
4.进程方法join
在上述创建进程代码中,父进程执行完毕之后才会执行子进程,其实两者并没有先后的关系,执行的顺序是操作系统来决定的,并且生成子进程也会消耗一段时间,故子进程在父进程之后,如果我们想要实现子进程完成之后在执行父进程,则需要join方法
1 from multiprocessing import Process 2 import time 3 4 def run(msg,i): 5 print('%s开始'%msg) 6 time.sleep(i) 7 print('%s结束'%msg) 8 9 if __name__ == '__main__': 10 start = time.time() # 记录开始时间 11 p = Process(target=run,args=('子进程',0)) 12 p1 = Process(target=run,args=('子进程1',1)) 13 p2 = Process(target=run,args=('子进程2',2)) 14 p3 = Process(target=run,args=('子进程3',3)) 15 p.start() 16 p1.start() 17 p2.start() 18 p3.start() 19 p.join() # 加入该方法后子进程会先执行完毕 20 p1.join() 21 p2.join() 22 p3.join() 23 end = time.time() # 记录结束时间 24 print('父进程') 25 print(end - start) # 打印总共的运行时间
join方法会等待所有子进程结束后再结束父进程,但是各个进程之间是互相独立的,他们等待的时间都在运行着,所以最后的运行时间是按照最长的等待时间加上真正的运行时间
5.进程间数据隔离问题
1 from multiprocessing import Process 2 age = 18 3 4 def run(): 5 global age 6 age = 28 7 print('子进程的age:%s'%age) 8 9 if __name__ == '__main__': 10 p = Process(target=run) 11 p.start() 12 p.join() 13 print('主进程的age:%s' % age)
子进程更改了age,主进程中的age并没有跟着改变
6.进程对象及其他方法
os.getpid:可以查看当前进程的pid(进程id)
os.getppid:可以查看当前进程的父进程的pid(进程id)
current_process模块:可以查看当前进程的pid(进程id),和os.getpid一样
p.terminate:强制终止子进程p
p.is_alive():判断子进程p是否还存活
1 from multiprocessing import Process,current_process 2 import time 3 import os 4 5 def run(): 6 print('os模块的子进程的pid:%s'%os.getpid()) 7 print('current_process模块子进程的pid:%s'%current_process()) 8 time.sleep(1) 9 print('父进程的pid:%s'%os.getppid()) 10 11 if __name__ == '__main__': 12 p = Process(target=run) 13 p.start() 14 time.sleep(0.5) 15 p.terminate() # 杀死一个子进程,一旦执行到这句话,不管子进程中的代码执行到哪了,都会立刻结束子进程 16 print(p.is_alive()) # 判断子进程是否存活,是个bool类型的值 17 print('主进程的pid:%s' % os.getpid()) 18 print('主进程的父进程的pid:%s' % os.getppid())
7.僵尸进程和孤儿进程
僵尸进程就是子进程死亡后,父进程还存放着子进程的pid之类的信息,一旦僵尸进程过多,会占用系统资源
父进程回收死亡的子进程(僵尸进程)资源的两种方式
1.join方法
2.父进程正常死亡
注意:所有进程都会步入僵尸进程
孤儿进程就是子进程没死,父进程意外死亡
针对linux会有儿童福利院(init),如果父进程意外死亡他所创建的子进程都会被福利院收养
8.守护进程
守护进程会随着主进程的结束而结束
主进程创建守护进程
守护进程会在主进程代码执行结束后就终止
守护进程内无法再开启子进程,否则就会抛出异常
注意:进程之间是互相独立的,主进程代码运行结束,守护进程随即终止
1 from multiprocessing import Process 2 import time 3 4 def run(msg): 5 print('%s开始'%msg) 6 time.sleep(1) 7 print('%s结束'%msg) 8 9 if __name__ == '__main__': 10 p = Process(target=run,args=('子进程',)) 11 p.daemon = True # 默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置 12 p.start() 13 time.sleep(1) 14 print('主进程结束')
8.互斥锁
一个简易的没有互斥锁的抢票软件
1 import json 2 from multiprocessing import Process 3 import time 4 5 # 查票 6 def search(i): 7 with open('ticket','r',encoding='utf-8') as f: 8 json_dict = f.read() 9 dict = json.loads(json_dict) 10 print('用户%s余票还有%s'%(i,dict.get('ticket'))) 11 12 # 买票 13 def buy(i): 14 with open('ticket','r',encoding='utf-8') as f: 15 json_dict = f.read() 16 dict = json.loads(json_dict) 17 time.sleep(1) 18 if dict.get('ticket') > 0 : 19 dict['ticket'] -= 1 20 with open('ticket','w', encoding='utf-8')as f: 21 d = json.dumps(dict) 22 f.write(d) 23 print('用户%s抢票成功'%i) 24 else: 25 print('用户%s卖完了'%i) 26 def run(i): 27 search(i) 28 buy(i) 29 30 if __name__ == '__main__': 31 for i in range(1,10): 32 p = Process(target=run,args=(i,)) 33 p.start()
当余票为1的时候,每个人都抢到了,这显然是不被允许的
这时候就要加入互斥锁
互斥锁:
将并发变成串行,虽然降低了效率但是提高了数据的安全
注意:
1.不要轻易的使用锁,容易造成死锁现象
2.只在处理数据的部分加锁,不要在全局加锁
3.锁必须在主进程中产生,不要在全局加锁
一个简易的加入了互斥锁的抢票系统
1 import json 2 from multiprocessing import Process,Lock 3 import time 4 5 # 查票 6 def search(i): 7 with open('ticket','r',encoding='utf-8') as f: 8 json_dict = f.read() 9 dict = json.loads(json_dict) 10 print('用户%s余票还有%s'%(i,dict.get('ticket'))) 11 12 # 买票 13 def buy(i): 14 with open('ticket','r',encoding='utf-8') as f: 15 json_dict = f.read() 16 dict = json.loads(json_dict) 17 time.sleep(1) 18 if dict.get('ticket') > 0 : 19 dict['ticket'] -= 1 20 with open('ticket','w', encoding='utf-8')as f: 21 d = json.dumps(dict) 22 f.write(d) 23 print('用户%s抢票成功'%i) 24 else: 25 print('用户%s卖完了'%i) 26 27 def run(i,mutex): 28 search(i) 29 mutex.acquire() # 抢锁,有人抢到了这把锁,其他人都要等他用完才能再抢 30 buy(i) 31 mutex.release() # 释放锁 32 33 if __name__ == '__main__': 34 mutex = Lock() # 在主进程中生成一把锁 35 for i in range(1,5): 36 p = Process(target=run,args=(i,mutex)) 37 p.start()
9.进程间通信
1.队列
进程间通信通过创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue来实现多进程之间的数据传递
Queue([maxsize])
通过队列来创建
maxsize是队列中允许的最大项数,省略的话就没有大小限制
底层是使用管道和锁定实现的
队列的几种方法:
q.put(参数)
将传入的参数放入队列,如果队列已满,则处于阻塞态,直到有空间为止
q.get()
从队列q中取出一个参数,如果队列为空,则处于阻塞态,直到队列中有值可以取出为止
q.get_nowait()
从队列中取值,如果没有值直接报错
q.full()
判断队列是否已满,满返回True
q.empty()
判断队列是否为空,空就返回True
1 from multiprocessing import Queue 2 3 4 q = Queue(3) # 队列限制大小为3 5 q.put(1) 6 q.put(2) 7 q.put(3) 8 # p.put(4) # 此时传入第四个参数会处于阻塞态 9 print(q.full()) # 判断队列是否已满,满就返回True 10 print(q.get()) 11 print(q.get()) 12 print(q.get()) 13 # print(p.get()) # 此时获取第三个参数会处于阻塞态 14 print(q.empty()) # 判断队列是否为空,如果为空就返回True 15 # q.get_nowait() # 取值,没有值直接报错
注意:上面的几种方法,除了传值和取值之外,都不适合在多进程中使用,因为子进程和主进程是两个不同执行的个体,不能用上面的几种方法草草的判断.
2.进程间通信IPC机制
把队列当做一个中间的空间,让多个进程在队列中实现数据的交换,从而实现通信
1 from multiprocessing import Process,Queue 2 3 def producer(q): 4 q.put('哈哈哈') # 传值 5 print('给队列传入值') 6 7 def consumer(q): 8 print(q.get()) # 取值 9 print('从队列取出值') 10 11 12 if __name__ == '__main__': 13 q = Queue() # 队列实例化对象 14 p1 = Process(target=producer,args=(q,)) # 生成两个子进程 15 p2 = Process(target=consumer,args=(q,)) 16 p1.start() 17 p2.start()
3.生产者消费者模型
1 from multiprocessing import Queue,Process 2 import time 3 import random 4 5 def producer(q,name,food): 6 for i in range(random.randint(5,10)): 7 time.sleep(random.random()) 8 data = '%s做了%s%s'%(name,food,i) 9 q.put(data) 10 11 12 def consumer(q,name): 13 while True: 14 time.sleep(random.random()) 15 data = q.get() 16 print('%s吃了%s'%(name,data)) 17 18 19 if __name__ == '__main__': 20 q = Queue() # 实例化队列对象 21 p1 = Process(target=producer,args=(q,'sxc','烧麦')) # 生成两个生存者 22 p2 = Process(target=producer,args=(q,'zzj','黄焖鸡')) 23 c1 = Process(target=consumer,args=(q,'zzp')) # 生成两个消费者 24 c2 = Process(target=consumer,args=(q,'lzx')) 25 p1.start() 26 p2.start() 27 c1.start() 28 c2.start() # 如果不加任何结束条件,此时会处于无限等待状态,即队列已被取完,但是还在等待传值from multiprocessing import Queue,Process 29 import time 30 import random 31 32 def producer(q,name,food): 33 for i in range(random.randint(5,10)): 34 time.sleep(random.random()) 35 data = '%s做了%s%s'%(name,food,i) 36 q.put(data) 37 38 39 def consumer(q,name): 40 while True: 41 time.sleep(random.random()) 42 data = q.get() 43 print('%s吃了%s'%(name,data)) 44 45 46 if __name__ == '__main__': 47 q = Queue() # 实例化队列对象 48 p1 = Process(target=producer,args=(q,'sxc','烧麦')) # 生成两个生存者 49 p2 = Process(target=producer,args=(q,'zzj','黄焖鸡')) 50 c1 = Process(target=consumer,args=(q,'zzp')) # 生成两个消费者 51 c2 = Process(target=consumer,args=(q,'lzx')) 52 p1.start() 53 p2.start() 54 c1.start() 55 c2.start() # 如果不加任何结束条件,此时会处于无限等待状态,即队列已被取完,但是还在等待传值
此时的问题是消费者取完值后会处于等待取值的状态
我们可以通过手动添加结束条件的方法
1 from multiprocessing import Queue,Process 2 import time 3 import random 4 5 def producer(q,name,food): 6 for i in range(random.randint(5,10)): 7 time.sleep(random.random()) 8 data = '%s做了%s%s'%(name,food,i) 9 q.put(data) 10 11 12 def consumer(q,name): 13 while True: 14 time.sleep(random.random()) 15 data = q.get() 16 if data == None:break 17 print('%s吃了%s'%(name,data)) 18 19 20 if __name__ == '__main__': 21 q = Queue() # 实例化队列对象 22 p1 = Process(target=producer,args=(q,'sxc','烧麦')) # 生成两个生存者 23 p2 = Process(target=producer,args=(q,'zzj','黄焖鸡')) 24 c1 = Process(target=consumer,args=(q,'zzp')) # 生成两个消费者 25 c2 = Process(target=consumer,args=(q,'lzx')) 26 p1.start() 27 p2.start() 28 c1.start() 29 c2.start() # 如果不加任何结束条件,此时会处于无限等待状态,即队列已被取完,但是还在等待传值 30 p1.join() 31 p2.join() # 子进程结束确保队列中所有的值都传完 32 q.put(None) # 通过向队列中传None的方式结束循环 33 q.put(None) # 问题是有几个消费者我们就要传几个None,这样做很不科学
此时的问题是我们不知道消费者有几个,有几个消费者就要添加几None,这样做很傻
引入一个JoinableQueue可连接的共享队列
这也是一个队列对象,但是他允许使用者通过生产者项目已成功处理,通知进程是使用共享的信号和条件变量来实现的
1 from multiprocessing import JoinableQueue,Process 2 import time 3 import random 4 5 def producer(q,name,food): 6 for i in range(random.randint(5,10)): 7 time.sleep(random.random()) 8 data = '%s做了%s%s'%(name,food,i) 9 q.put(data) 10 11 12 def consumer(q,name): 13 while True: 14 time.sleep(random.random()) 15 data = q.get() 16 # if data == None:break 17 print('%s吃了%s'%(name,data)) 18 q.task_done() # 发出信号,表示get的值已被处理完毕 19 20 21 if __name__ == '__main__': 22 q = JoinableQueue() # 实例化队列对象 23 p1 = Process(target=producer,args=(q,'sxc','烧麦')) # 生成两个生存者 24 p2 = Process(target=producer,args=(q,'zzj','黄焖鸡')) 25 c1 = Process(target=consumer,args=(q,'zzp')) # 生成两个消费者 26 c2 = Process(target=consumer,args=(q,'lzx')) 27 p1.start() 28 p2.start() 29 c1.daemon = True # 将消费者设置为守护进程 30 c2.daemon = True 31 c1.start() 32 c2.start() # 如果不加任何结束条件,此时会处于无限等待状态,即队列已被取完,但是还在等待传值 33 p1.join() 34 p2.join() # 子进程结束确保队列中所有的值都传完 35 # q.put(None) # 通过向队列中传None的方式结束循环 36 # q.put(None) # 问题是有几个消费者我们就要传几个None,这样做很不科学 37 q.join() # 等待队列中的数据全被取出
30