一、进程理论
1、程序和进程
程序：一堆代码
进程：正在运行的程序
进程是一个抽象的概念，每一个进程都有它自己独立的内存空间
2、同步和异步：针对任务的提交方式
同步：提交任务之后，原地等待任务的返回结果，期间不做任何事（叫人吃饭，一直等待）
异步：提交任务之后，不等待任务的返回结果，执行运行下一行代码（叫人吃饭，但自己先走）
3、阻塞与非阻塞：针对程序运行的状态
阻塞：遇到io操作 ->>阻塞态
非阻塞：就绪或者运行态
二、创建进程的两种方式
1、方式一
from multiprocessing import Process import time def task(name): print('%s is running' % name) time.sleep(3) print('%s is over' % name) # 注意，在windows系统中，创建进程会将代码以模块的方式从头到尾加载一遍 # 一定要写在if __name__ == '__main__':代码块里面 # 强调:函数名一旦加括号，执行优先级最高，立刻执行 if __name__ == '__main__': p1 = Process(target=task, args=('egon',)) # # 这一句话只是实例化了一个Process对象 p1.start() # 告诉操作系统创建一个进程，操作系统会申请内存空间，然后把父进程的数据拷贝给子进程，作为子进程的初始状态 print('主进程')
2、方式二
from multiprocessing import Process import time class MyProcess(Process): def __init__(self, name): super().__init__() self.name = name def run(self): print('%s is running' % self.name) time.sleep(1) print('%s is end' % self.name) if __name__ == '__main__': obj = MyProcess('egon') obj.start() print('主进程')
三、join方法：join主进程等待子进程结束才继续运行
from multiprocessing import Process import time def task(name, n): print('%s is running' % name) time.sleep(n)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) # 结果为 子进程0 is running 子进程1 is running 子进程2 is running 主进程 2.624150037765503
四、进程对象其他方法
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(0.1) print(p1.is_alive()) # 判断子进程是否存活 print('主')
五、进程之间内存隔离
# 记住进程与进程之间数据是隔离的！！！ from multiprocessing import Process x = 100 def task(): global x x = 20 print('子进程', x) if __name__ == '__main__': p = Process(target=task) p.start() p.join() print('主进程', x) # 结果为 子进程 20 主进程 100
六、守护进程
主进程一旦运行完毕，子进程立即结束运行
应用场景：开子进程的目的是并发的执行任务，当这个任务是伴随着主进程整个生命周期结束之后，这个任务也就没有存在的意义了，就应该把子进程做成守护进程
from multiprocessing import Process import time def task(name): print('%s正活着' % name) time.sleep(3) if __name__ == '__main__': p = Process(target=task, args=('egon总管',)) p.daemon = True p.start() print('主进程正在结束')
解果为
主进程正在结束
七、互斥锁
①锁千万不要随意去用 ②牺牲了效率但是保证了数据的安全 ③锁一定要在主进程中创建，给子进程去用 ④解决多个进程操作同一份数据造成数据不安全的情况 ⑤加锁会将并发变成串行 ⑥锁通常用在对数据操作的部分，并不是对进程全程加锁 mutex.acquire() # 抢锁 一把锁不能同时被多个人使用,没有抢到的人，就一直等待锁释放 buy(i) mutex.release() # 释放锁
from multiprocessing import Process, Lock import json import time import random 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查询余票为0' % i) def run(i, mutex): search(i) mutex.acquire() # 抢锁 一把锁不能同时被多个人使用,没有抢到的人，就一直等待锁释放 buy(i) mutex.release() # 释放锁 if __name__ == '__main__': mutex = Lock() # 创建锁 for i in range(5): p = Process(target=run, args=(i, mutex)) p.start() # 结果为 用户查询余票数3 用户查询余票数3 用户查询余票数3 用户查询余票数3 用户查询余票数3 用户1抢票成功 用户0抢票成功 用户2抢票成功 用户3查询余票为0 用户4查询余票为0
八、进程间通信 IPC（InterProcess Communication）
1、队列：先进先出，堆栈：先进后出
注意：队列用来储存进程之间沟通的消息，数据量不应该过大
maxsize的值超过内存限制就变的毫无意义。q = Queue(maxsze)
from multiprocessing import Queue q = Queue(5) # 产生一个最多能够存放五个数据的队列 q.put(1) q.put(2) q.put(3) print(q.full()) # False q.put(4) q.put(5) # 只能放进去五个 print(q.full()) # True # 判断队列是否放满 q.put(6) # 再放一个放不进去了，程序不会报错会阻塞 # 存储数据 for i in range(5): q.put(i) # for循环往队列里面存放数据 print(q.get()) # 0 # 取数据，get一次就取一个 print(q.get()) # 1 print(q.get()) # 2 print(q.empty()) # False print(q.get()) # 3 print(q.get()) # 4 print(q.empty()) # True # 判断队列数据是否为空，需要注意的是，在并发的情况下，这个方法判断不准确！ print(q.get()) # 取完了，队列为空，get会在原地等待队列中有数据过来
2、基于队列实现进程间通信
from multiprocessing import Queue, Process def producer(q): q.put('你好') def consumer(q): print(q.get()) if __name__ == '__main__': q = Queue() # 生成一个队列对象 p1 = Process(target=producer, args=(q,)) c1 = Process(target=consumer, args=(q,)) p1.start() c1.start() # 结果为 你好
九、生产者消费者模型（*****）
1、生产者：生产数据的任务 消费者：处理数据的任务 2、两者之间的通信介质：队列 / 管道 3、为什么有： 解决供需不平衡的问题 4、怎么解决： 定义一个队列，用来存放固定数量的数据 生产者生产的数据放到队列里面，消费者去队列里面获取数据 5、实现生产者消费者模型三要素 生产者 消费者 队列 6、什么时候用该模型： 程序中出现明显的两类任务，一类任务是负责生产数据，另外一类任务是负责处理生产的数据的 7、该模型的好处： ①实现了生产者与消费者解耦和 ②平衡了生产力与消费力，即生产者可以一直不停地生产，消费者可以不停地处理，因为二者不再直接沟通的，而是跟队列沟通
from multiprocessing import Process, JoinableQueue import time def producer(name, food, q): for i in range(5): data = '%s生产了%s %s' % (name, food, i) time.sleep(2) print(data) q.put(data) # 将生产的数据放入队列中 def consumer(name, q): while True: data = q.get() if data is None: break time.sleep(2) print('%s吃了%s' %(name, data)) q.task_done() # 告诉你的队列，你已经将数据取出并且处理完毕 if __name__ == '__main__':
# JoinableQueue: 可join的队列，该队列相比普通的Queue的区别在于该队列额外增加了join函数 q = JoinableQueue() # 生成一个队列对象 p1 = Process(target=producer, args=('egon', '包子', q)) p2 = Process(target=producer, args=('tank', '油条', q)) c1 = Process(target=consumer, args=('jerry', q)) c2 = Process(target=consumer, args=('tom', q)) p1.start() p2.start() c1.daemon = True # c.daemon = True: 主进程运行完不会检查子进程的状态（是否执行完），直接结束进程； c2.daemon = True # 子进程跟随主进程结束而结束 c1.start() c2.start() # 等待生产者生产完所有的数据 p1.join() p2.join() # 在生产者生产完数据之后，往队列里面放一个提示性的消息，告诉消费者已经没有数据，你走吧，不要等了 q.join() # q.join()等待所有数据全部处理完毕 print('主')