并发编程——多进程
1、什么是进程？
程序：堆代码
进程：正在运行的程序
进程是一个实体，每个进程都自己的独立内存空间
进程的三个状态：
multiprocessing模块介绍
2、同步和异步：针对的程序运行的状态
同步：提交任务后原地等待任务返回结果，期间不做任何事情
异步：提交任务后，不等待任务返回结果，直接运行下一行代码
3、阻塞和非阻塞：针对程序运行的状态
阻塞：遇到 IO 操作》》》阻塞状态
非阻塞：就绪或者运行状态 >>>>就绪状态
multiprocessing模块介绍
Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动） 强调： 1. 需要使用关键字的方式来指定参数 2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号
参数介绍：
1 group参数未使用，值始终为None
2
3 target表示调用对象，即子进程要执行的任务
4
5 args表示调用对象的位置参数元组，args=(1,2,'egon',)
6
7 kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}
8
9 name为子进程的名称
方法介绍： 1 p.start()：启动进程，并调用该子进程中的p.run() 2 p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法 3 4 p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁 5 p.is_alive():如果p仍然运行，返回True 6 7 p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
4、开启进程的两种方式：
form multiprocessing import Pocess import time def task(name) print('%s is running '%name) time.sleep(3) print('%s is over' %name) if __name__ == '__main__': p1=Process(target=task,args=()) p1.strat() print('主程序')方式1
from multiprocessing import Process import time class MyProcess(Process): def __init__(self, name): super().__init__() self.name = name #必须写run方法 def run(self): print('%s is running'%self.name) time.sleep(3) print('%s is over'%self.name) if __name__ == '__main__': obj = MyProcess('egon') obj.start() print('主')方式2
5、进程对象的join方法:
# join的作用仅仅只是让主进程等待子进程的结束，不会影响子进程的运行
from multiprocessing import Process import time def task(name, n): print('%s is running'%name) time.sleep(n) print('%s is over'%name) if __name__ == '__main__': start_time = time.time() p_list = [] for i in range(3): p = Process(target=task,args=('子进程%s'%i,i)) p.start() p_list.append(p) for i in p_list: i.join() print('主',time.time()-start_time)join
6、进程之间空间独立
from multiprocessing import Process x = 100 def task(): global x x = 1 print(x) if __name__ == '__main__': p = Process(target=task) p.start() p.join() print('主', x) ''' F:\python36\python3.exe F:/python_s7/week08/day02/进程间数据是隔离的.py 1 主 100 '''空间独立
7、进程对象其他相关方法
from multiprocessing import Process,current_process import time import os def task(): print('%s is running'%os.getpid()) time.sleep(3) print('%s is over'%os.getppid()) if __name__ == '__main__': p1 = Process(target=task) p1.start() # p1.terminate()# 杀死子程序 time.sleep(3) print(p1.is_alive())#判断子程序 print('主')其他方法
8、守护进程
主进程创建守护进程
其一：守护进程会在主进程代码执行结束后就终止
其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
from multiprocessing import Process import time def task(name): print('%s 正活着'%name) time.sleep(3) print('%s 正常死亡'%name) if __name__ == '__main__': p = Process(target=task,args=('egon总管',)) p.daemon = True # 必须在p.start开启进程命令之前声明 p.start() print('皇帝jason正在死亡')join
9、互斥锁：
牺牲了效率但是保证了数据的安全
锁一定要在主进程中创建，给子进程去用
解决多个进程操作同一份数据，造成数据不安全的情况
加锁会将并发变成串行
锁通常用在对数据操作的部分，并不是对进程全程加锁
mutex.acquire() # 抢锁 一把锁不能同时被多个人使用,没有抢到的人，就一直等待锁释放
buy(i)
mutex.release() # 释放锁
import json import time import random from multiprocessing import Process, Lock # 查票 def search(i): with open('info', 'r', encoding='utf-8')as f: data = json.load(f) print('用户查询余票%s' % data.get('ticket')) def buy(i): # 购买前想查询余票 with open('info', 'r', encoding='utf-8') as f: data = json.load(f) # 模拟延迟 time.sleep(random.randint(1, 3)) if data.get('ticket') > 0: data['ticket'] -= 1 with open('info', 'w', encoding='utf-8')as f: json.dump(data, f) print('用户抢票成功%s' % i) else: print('用户%s抢票失败' % i) def run(i, mutex): search(i) mutex.acquire() buy(i) mutex.release() if __name__ == '__main__': mutex = Lock() for i in range(10): p = Process(target=run, args=(i, mutex)) p.start()互斥锁
10 、进程之间的通讯（队列）
隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的
Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。
1 q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。 2 q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常. 3 4 q.get_nowait():同q.get(False) 5 q.put_nowait():同q.put(False) 6 7 q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。 8 q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。 9 q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样
例子：
from multiprocessing import Queue # 产生一个队列能放5个值 q = Queue(5) q.put(1) q.put(2) q.put(3) q.put(4) q.put(5) print(q.full()) # print(q.get()) # print(q.get()) # print(q.get()) # print(q.get()) # print(q.get()) # print(q.empty())# 判断队列是否为空，需要注意的是，在并发的情况下，这个方法判断不准确！ print(q.get_nowait())# 和get一样取值，但是当队列没有值可以取的时候回报错 print(q.get_nowait()) print(q.get_nowait()) print(q.get_nowait()) print(q.get_nowait())进程之间通讯
11、生产者与消费者模
"""
生产者消费者模型
生产者:做包子的 生产数据的
消费者:买包子的 处理数据的
解决供需不平衡的问题
定义一个队列，用来存放固定数量的数据
解决一个生产者和消费者不需直接打交道，两者都通过队列实现数据传输
Queue:管道+锁
"""
from multiprocessing import Queue,Process,JoinableQueue
import time
import random
def producer(name,food,q):
for i in range(5):
data = '%s生产了%s%s'%(name,food,i)
time.sleep(random.randint(1,3))
print(data)
q.put(data) # 将生产的数据放入队列中
def consumer(name,q):
while True:
data = q.get()
if data is None:break
time.sleep(random.randint(1, 3))
print('%s吃了%s'%(name,data))
q.task_done() # 告诉你的队列，你已经将数据取出并且处理完毕
if __name__ == '__main__':
q = JoinableQueue() # 生成一个队列对象
p1 = Process(target=producer,args=('大厨egon','包子',q))
p2 = Process(target=producer,args=('靓仔tank','生蚝',q))
c1 = Process(target=consumer,args=('吃货owen',q))
c2 = Process(target=consumer,args=('坑货kevin',q))
p1.start()
p2.start()
c1.daemon = True
c2.daemon = True
c1.start()
c2.start()
# 等待生产者生产完所有的数据
p1.join()
p2.join()
# 在生产者生产完数据之后，往队列里面放一个提示性的消息，告诉消费者已经没有数据，你走吧，不要等了
# q.put(None)
# q.put(None)
q.join() # 等待队列中数据全部取出
print('主')
优质内容筛选与推荐>>
1、2.NET Core设定数据库种子
2、采购订单单价异常控制
3、VS2008下安装OpenCV2.0
4、2019牛客多校第四场K number dp or 思维
5、课后作业03-验证课件上的代码，并将所有的动手动脑或要求发表博客作业部分整理成一篇博客