一：进程理论

1.程序和进程

程序就是一堆代码文件，是指令和数据的集合，可以作为目标文件保存在磁盘中，或者作为段存放在内存地址空间中。（静态）

进程就是一个程序运行的过程，是操作系统进行资源分配和保护的基本单位（动态）

进程是资源分配的最小单位

1个程序可以对应多个进程，但是1个进程只能对应1个程序。进程和程序的关系犹如演出和剧本的关系

2.并发、并行、串行

串行：多个任务依次运行、一个进程运行完毕、在运行下一个

并发：多个任务看起来像是在同时运行，本质上还是一个个地运行

并行：多个任务真正意义上一起运行，只有多个CPU才能并行的概念

Python并发编程-进程 - 图1

3.同步-异步/阻塞 - 非阻塞

同步（慢）：
发起一个请求，直到请求返回结果之后，才进行下一步操作

  1. 就行运行在一段代码，自上而下一行一行运行，只有上一行代码执行完成，才能执行下一行代码
  3. 简单来说：**同步**就是：**必须一件事一件事的做**，**等前一件事做完了，才能做下一件事**

异步（快）：

  1. **发起一个请求后，不管这个请求有没有返回结果，直接可以进行下一步操作**
  3. 一般情况下，有一个回调的操作
  5. 简单来说：**异步**就是：**可以多件事情同时做**。

阻塞（慢）：

  1. 遇到IO操作，CPU会被操作系统切换到其它进程。
  3. **调用结果返回之前，当前线程会被挂起**。调用线程只用在得到结果之后才会返回。
  5. 简单来说，**阻塞**就是：需要买奶茶和冰淇淋，买奶茶的时候奶茶制作过程**只能等**，然后买完奶茶再去买冰淇淋

非阻塞（快）：

  1. 没有IO操作，一直运行
  3. **在不能立刻得到结果之前，改调用不会阻塞当前线程**
  5. 简单来说，**非阻塞**就是：需要买奶茶和冰淇淋，买奶茶的时候等的时间去买冰淇淋。

最佳状态：异步+非阻塞
拓展：

  1. 同步框架：Flak、Django3.0之前
  3. 异步框架：Tornado、Sanic、FastAPI

例子：

背景：妈妈出门了，小明想要看电视，但是妈妈让小明烧水。

1.小明烧水，在一旁等待，时不时看一下水有没有烧开。 —— 同步阻塞

2.小明烧水，等待的时间偷偷去看会儿电视，看电视的时候时不时来看一下水有没有烧开。 —— 同步非阻塞

3.小明买了一个水开了之后会响的水壶来烧水，烧水时在一旁等着，不用时不时看水有木有烧开。 —— 异步阻塞

4.小明用那个会响的水壶烧水，然后去看电视，等水开了之后发出响声，再去开。 —— 异步非阻塞

4.进程运行的三种状态

打开一个应用程序，该进程会进入就绪态，获取CPU资源后，执行程序代码，进入运行态，出现了读写文件with open(…)，就会进入阻塞态，CPU资源就会被操作系统分配到其他进程

运行态（Running）(非阻塞)：进程已获CPU，正在执行。单处理机系统中。处于执行状态的进程只有一个；多处理机系统中，有多个处于执行状态的进程。

就绪态（Ready）（非阻塞）：进程以获得除CPU外的所有必要资源，只等待CPU时的状态。一个系统会将多个处于就绪状态的进程排成一个就绪队列

阻塞态（Blocked）（阻塞）：正在执行的进程由于某种原因（文件读写等IO操作）而暂时无法继续执行，便放弃处理机而处于暂停状态，即进程执行受阻。（这种状态又称等待状态或封锁状态）

就绪 → 运行

处于就绪状态的进程，当进程调度程序为之分配了处理器后，该进程便由就绪状态转变成运行状态。

运行 → 就绪

处于执行状态的进程在其执行过程中，因分配给它的一个时间片已用完而不得不让出处理机，于是进程从运行状态转变成就绪状态。

运行 → 阻塞

正在执行的进程因等待某种事件(文件读写等IO操作)发生而无法继续执行时，便从运行状态变成阻塞状态。

阻塞 → 就绪

处于阻塞状态的进程，若其等待的事件已经发生，于是进程由阻塞状态转变为就绪状态。

Python并发编程-进程 - 图2

优化程序效率的核心法则

降低IO操作（硬盘IO、网络IO）

数据获取优先级：

内存 > 本地硬盘 > 网络IO

5.提交任务的2种方式

同步调用：一个任务必须要执行完毕才能执行另一个任务

异步调用：一个任务在执行过程中，可以执行另一个任务

二：进程的使用

　１.进程的创建

Windows：CreateProcess

Linux：Fork

2.进程的终止

1.正常退出（自愿；程序执行完毕后，终止，资源被回收）

2.出错退出（自愿；python3 test.py但是test.py不存在）

3.严重错误（非自愿；执行非法指令）

4.被其他进程杀死（非自愿；被操作系统杀死taskkill /F /PID 3333）

三：join方法

0.作死法

  1. import time
  2. from multiprocessing import Process
  5. def task(n):
  6.     print('%s run' % n)
  7.     time.sleep(10000)
  10. if __name__ == '__main__':
  11.     for i in range(10000):
  12.         p = Process(target=task, args=(1,))
  13.         p.start()
  14.     print('主进程')

1.单个进程 + 单个子进程

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task(n):
  7.     print(f'task[{os.getpid()}] is running')
  8.     time.sleep(n)
  9.     print(f'task[{os.getpid()}] is done')
  12. if __name__ == '__main__':
  13.     # p = Process(target=task, args=(5,))       # args中为元组/列表
  14.     p = Process(target=task, kwargs={'n': 5})   # kwargs中为字典
  15.     p.start()  # 在向发操作系统发送启动子进程的信号，属于IO操作，速度慢
  16.     print(f'主进程[{os.getpid()}]')
  18. # 主进程[15188]
  19. # task[86956] is running
  20. # task[86956] is done

2.单个父进程 + 多个子进程 方式1

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task(n, name):
  7.     print(f'task[{name}] is running [{os.getpid()}]')
  8.     time.sleep(n)
  9.     print(f'task[{name}] is done [{os.getpid()}]')
  12. if __name__ == '__main__':
  13.     p1 = Process(target=task, args=(2, 'p1'))
  14.     p2 = Process(target=task, args=(2, 'p2'))
  15.     p3 = Process(target=task, args=(2, 'p3'))
  17.     p1.start()  # 在向发操作系统发送启动子进程的信号，属于IO操作，速度慢
  18.     p2.start()  # 在向发操作系统发送启动子进程的信号，属于IO操作，速度慢
  19.     p3.start()  # 在向发操作系统发送启动子进程的信号，属于IO操作，速度慢
  21.     print(f'主进程[{os.getpid()}]')
  24. # 主进程[79800]
  25. # task[p1] is running [17320]
  26. # task[p3] is running [66276]
  27. # task[p2] is running [36784]
  28. # task[p1] is done [17320]
  29. # task[p3] is done [66276]
  30. # task[p2] is done [36784]

3.单个父进程 + 多个子进程 方式2

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. class Myprocess(Process):
  7.     def __init__(self, tag):
  8.         super().__init__()
  9.         self.tag = tag
  11.     def run(self) -> None:    # 表示run函数的返回值为空，无返回值
  12.         print(f'task[{self.tag}] is running [{os.getpid()}]')
  13.         time.sleep(3)
  14.         print(f'task[{self.tag}] is done [{os.getpid()}]')
  17. if __name__ == '__main__':
  18.     p1 = Myprocess('进程1')
  19.     p2 = Myprocess('进程2')
  20.     p3 = Myprocess('进程3')
  22.     p1.start()  # 相当于：p1.run，但是p1.run是同步的，所以这里用p1.start
  23.     p2.start()  # p2.run
  24.     p3.start()  # p3.run
  26.     print(f'主进程[{os.getpid()}]')

4.进程之间的数据是隔离的

  1. 因为：**开启1个进程，就是开启了1个Python解释器。**
  3. **Windows**下要开进程，需要在**main函数**下
  1. import os
  2. import time
  3. from multiprocessing import Process
  5. age = 18
  7. def task(n, name):
  8.     global age  # 局部修改全局
  9.     age = 99
  10.     print(f'task[{name}] is running [{os.getpid()}]')
  11.     time.sleep(n)
  12.     print(f'task[{name}] is done [{os.getpid()}]')
  13.     print(f'子进程的age：{age}')
  16. if __name__ == '__main__':
  17.     p1 = Process(target=task, args=(1, 'p1'))
  18.     p1.start()
  19.     print(f'主进程[{os.getpid()}]')
  20.     print(f'主进程的age：{age}')
  22. # 主进程[2680]
  23. # 主进程的age：18
  24. # task[p1] is running [45116]
  25. # task[p1] is done [45116]
  26. # 子进程的age：99

5.join - 让子进程的开启者 等p1开启并运行完毕后 再运行

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task(n, name):
  7.     print(f'task[{name}] is running [{os.getpid()}]')
  8.     time.sleep(n)
  9.     print(f'task[{name}] is done [{os.getpid()}]')
  12. if __name__ == '__main__':
  13.     p1 = Process(target=task, args=(2, 'p1'))
  14.     p1.start()  # 在向发操作系统发送启动子进程的信号，属于IO操作，速度慢
  15.     p1.join()   # join让子进程执行完成
  16.     print(f'主进程[{os.getpid()}]')
  19. # task[p1] is running [47164]
  20. # task[p1] is done [47164]
  21. # 主进程[44356]

不用join - 不等待 耗时0.几秒

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task(n, name):
  7.     print(f'task[{name}] is running [{os.getpid()}]')
  8.     time.sleep(n)
  9.     print(f'task[{name}] is done [{os.getpid()}]')
  12. if __name__ == '__main__':
  13.     ctime = time.time()
  14.     p1 = Process(target=task, args=(1, 'p1'))
  15.     p2 = Process(target=task, args=(2, 'p1'))
  16.     p1.start()
  17.     p2.start()
  18.     # p1.join()
  19.     # p2.join()
  20.     print(f'主进程[{os.getpid()}]')
  21.     print(time.time() - ctime)
  23. # 不用join
  24. # 主进程[59116]
  25. # 0.020940065383911133
  26. # task[p1] is running [89420]
  27. # task[p1] is running [91328]
  28. # task[p1] is done [89420]
  29. # task[p1] is done [91328]

只用p1.join - 等待p1执行完毕再执行主进程

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task(n, name):
  7.     print(f'task[{name}] is running [{os.getpid()}]')
  8.     time.sleep(n)
  9.     print(f'task[{name}] is done [{os.getpid()}]')
  12. if __name__ == '__main__':
  13.     ctime = time.time()
  14.     p1 = Process(target=task, args=(1, 'p1'))
  15.     p2 = Process(target=task, args=(2, 'p2'))
  16.     p1.start()
  17.     p2.start()
  18.     p1.join()
  19.     # p2.join()
  20.     print(f'主进程[{os.getpid()}]')
  21.     print(time.time() - ctime)
  23. # 只用p1.join
  24. # task[p1] is running [61892]
  25. # task[p2] is running [38660]
  26. # task[p1] is done [61892]
  27. # 主进程[47172]
  28. # 1.1537489891052246
  29. # task[p2] is done [38660]

只用p2.join - 等待p2执行完毕再执行主进程

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task(n, name):
  7.     print(f'task[{name}] is running [{os.getpid()}]')
  8.     time.sleep(n)
  9.     print(f'task[{name}] is done [{os.getpid()}]')
  12. if __name__ == '__main__':
  13.     ctime = time.time()
  14.     p1 = Process(target=task, args=(1, 'p1'))
  15.     p2 = Process(target=task, args=(2, 'p2'))
  16.     p1.start()
  17.     p2.start()
  18.     # p1.join()
  19.     p2.join()
  20.     print(f'主进程[{os.getpid()}]')
  21.     print(time.time() - ctime)
  23. # 只用p2.join
  24. # task[p1] is running [56056]
  25. # task[p2] is running [47384]
  26. # task[p1] is done [56056]
  27. # task[p2] is done [47384]
  28. # 主进程[3424]
  29. # 2.159543514251709

同时p1.join和p2.join - 时间取决于最长的那个

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task(n, name):
  7.     print(f'task[{name}] is running [{os.getpid()}]')
  8.     time.sleep(n)
  9.     print(f'task[{name}] is done [{os.getpid()}]')
  12. if __name__ == '__main__':
  13.     ctime = time.time()
  14.     p1 = Process(target=task, args=(1, 'p1'))
  15.     p2 = Process(target=task, args=(2, 'p2'))
  16.     p1.start()
  17.     p2.start()
  18.     p1.join()
  19.     p2.join()
  20.     print(f'主进程[{os.getpid()}]')
  21.     print(time.time() - ctime)
  23. # p1.join + p2.join
  24. # task[p1] is running [56056]
  25. # task[p2] is running [47384]
  26. # task[p1] is done [56056]
  27. # task[p2] is done [47384]
  28. # 主进程[3424]
  29. # 2.159543514251709

6.join - 运行多个子进程 - 串行

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task(n, name):
  7.     print(f'task[{name}] is running [{os.getpid()}]')
  8.     time.sleep(n)
  9.     print(f'task[{name}] is done [{os.getpid()}]')
  12. if __name__ == '__main__':
  13.     p1 = Process(target=task, args=(2, 'p1'))
  14.     p2 = Process(target=task, args=(3, 'p2'))
  15.     p3 = Process(target=task, args=(4, 'p3'))
  17.     start = time.time()
  19.     p1.start()
  20.     p1.join()
  22.     p2.start()
  23.     p2.join()
  25.     p3.start()
  26.     p3.join()
  28.     print(f'主进程[{os.getpid()}]')
  29.     print('耗时：', time.time() - start)
  31. # task[p1] is running [72452]
  32. # task[p1] is done [72452]
  33. # task[p2] is running [31540]
  34. # task[p2] is done [31540]
  35. # task[p3] is running [62568]
  36. # task[p3] is done [62568]
  37. # 主进程[11548]
  38. # 耗时： 9.440884351730347

7.join - 运行多个子进程 - 并发

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task(n, name):
  7.     print(f'task[{name}] is running [{os.getpid()}]')
  8.     time.sleep(n)
  9.     print(f'task[{name}] is done [{os.getpid()}]')
  12. if __name__ == '__main__':
  13.     p1 = Process(target=task, args=(2, 'p1'))
  14.     p2 = Process(target=task, args=(3, 'p2'))
  15.     p3 = Process(target=task, args=(4, 'p3'))
  17.     start = time.time()
  19.     p1.start()
  20.     p2.start()
  21.     p3.start()
  23.     p1.join()    # 此时，p1、p2、p3的join顺序已经不重要了
  24.     p2.join()
  25.     p3.join()
  27.     print(f'主进程[{os.getpid()}]')
  28.     print('耗时：', time.time() - start)
  30. # task[p2] is running [47248]
  31. # task[p3] is running [46152]
  32. # task[p1] is running [43612]
  33. # task[p1] is done [43612]
  34. # task[p2] is done [47248]
  35. # task[p3] is done [46152]
  36. # 主进程[3776]
  37. # 耗时： 4.173335313796997

四：进程调度

想要多个进程交替运行，就需要操作系统来调度CPU的时间片

1.先来先服务（FCFS）(First Come First Serve)

先来先服务（FCFS）调度算法是一种最简单的调度算法，该算法既可用于作业调度，也可用于进程调度。

FCFS算法比较有利于长作业（进程），而不利于短作业（进程）。

由此可知，本算法适合于CPU繁忙型作业，而不利于I/O繁忙型的作业（进程）。

2.短作业优先

短作业（进程）优先调度算法（SJ/PF）是指对短作业或短进程优先调度的算法，该算法既可用于作业调度，也可用于进程调度。

但其对长作业不利；不能保证紧迫性作业（进程）被及时处理；作业的长短只是被估算出来的。

3.时间片轮转

时间片轮转(Round Robin，RR)法的基本思路是让每个进程在就绪队列中的等待时间与享受服务的时间成比例。

在时间片轮转法中，需要将CPU的处理时间分成固定大小的时间片，例如，几十毫秒至几百毫秒。

如果一个进程在被调度选中之后用完了系统规定的时间片，但又未完成要求的任务，则它自行释放自己所占有的CPU而排到就绪队列的末尾，等待下一次调度。

同时，进程调度程序又去调度当前就绪队列中的第一个进程。

4.多级反馈队列

前面介绍的各种用作进程调度的算法都有一定的局限性。

如短进程优先的调度算法，仅照顾了短进程而忽略了长进程，而且如果并未指明进程的长度，则短进程优先和基于进程长度的抢占式调度算法都将无法使用。

而多级反馈队列调度算法则不必事先知道各种进程所需的执行时间，而且还可以满足各种类型进程的需要，因而它是目前被公认的一种较好的进程调度算法。

五：僵尸进程 与 孤儿进程

1.僵尸进程

进程结束之后，还有数据没回收。

子进程结束后，父进程未将子进程资源回收，该子进程就编程了僵尸进程。

2.孤儿进程

主进程结束了，子进程还没结束。

父进程结束了，子进程未结束，该子进程变成了孤儿进程。

该子进程就会被专门的进程接管。

六：进程对象 及 其他方法

1.每一个进程 都会有自己的ID号，这个ID叫做PID

每个程序都可以有多个进程

Python并发编程-进程 - 图3

2.Windows中查看进程

  1. tasklist |findstr python

3.Linux、Mac中查看进程

  1. ps aux |grep

4.进程对象：

  1. ==p = Process(target = task, )==或者是在进程内部：current_process()

5.查看当前进程的ID：

p.pid、os.getpid()或者current_process().pid

6.获取父进程ID号：

os.getppid()，子进程中获取父进程ID号，等于父进程的ID号

7.查看当前进程是否存活：

p.is_alive()或者current_process().is_alive()

8.主进程的父进程

是pycharm,pycharm的父进程是系统（centos6中是init，centos7及以后是system）

9.关闭进程：

p.terminate()，只能在主进程中关闭子进程

七：守护进程

主进程结束之后，守护进程也随它而去，不会变成孤儿进程

进程名.daemon = True一定要加在进程名.start()之前

  1. import os
  2. import time
  3. from multiprocessing import Process
  6. def task():
  7.     print(f'子进程开始，ID为：{os.getpid()}')
  8.     time.sleep(100)
  9.     print('子进程结束')
  12. if __name__ == '__main__':
  13.     p = Process(target=task,)
  15.     # 守护进程：主进程一旦结束，子进程也结束
  16.     p.daemon = True     # 一定要加在启动之前，将p进程设置为主进程的守护进程
  17.     p.start()
  18.     time.sleep(1)
  19.     print('主进程')
  20.     print(f'主进程ID：{os.getpid()}')
  22. # 子进程开始，ID为：17324
  23. # 主进程
  24. # 主进程ID：66772

问题：

1.主进程的父进程是谁？

pycharm

2.主进程开了很多子进程，每个都需要设置守护进程吗？

根据需求来设置，如果需要，就设置：p.deamon = True

八：互斥锁

多个进程操作同一份数据的时候，会出现数据错乱的问题

针对上述问题，解决方式就是加锁处理：将并行变成串行，牺牲效率但是保证了数据的安全

json字符串转换网站：json.cn

多个人同时抢锁，只有一个人可以拿到，必须释放，其他人才能拿到。

例子：

10个人抢票，票只有1张，10个人可以同时查票，但是能买到票的只有1个，这个人买到票之后，其他的人就没票可以买了

ticket.txt

  1. {"ticket_count": 1}

mylock.py

  1. import json
  2. import time
  3. import random
  4. from multiprocessing import Process, Lock
  7. # 查票
  8. def search():
  9.     # 打开文件，读出ticket_couont
  10.     with open('ticket.txt', mode='r', encoding='UTF-8') as f:
  11.         dic = json.load(f)
  12.         # print(dic['ticket'])
  13.         print(f'剩余票量：{dic.get("ticket_count")}')
  16. # 买票
  17. def buy():
  18.     with open('ticket.txt', mode='r', encoding='UTF-8') as f:
  19.         dic = json.load(f)
  20.     time.sleep(random.randint(1, 3))  # 模拟网络延迟
  21.     if dic.get('ticket_count') > 0:  # # 余票大于0，能够买票
  22.         dic['ticket_count'] -= 1
  23.         # 保存到文件中
  24.         with open('ticket.txt', mode='w', encoding='UTF-8') as f:
  25.             json.dump(dic, f)
  26.             print('买票成功！')
  27.     else:
  28.         print('余票不足，买票失败！')
  31. # 写一个函数，先查票，再买票
  32. def task(mutex):
  33.     search()
  34.     # 买票过程 需要加锁
  35.     # 买前加锁
  36.     # mutex.acquire()
  37.     # buy()   # 10个进程，编程了串行
  38.     # # 买后释放锁
  39.     # mutex.release()
  40.     with mutex:
  41.         buy()
  44. if __name__ == '__main__':
  45.     # 锁的创建 应该设置在主进程中
  46.     mutex = Lock()  # 创建一把锁
  48.     # 模拟10个人查票+买票（开10个进程）
  49.     for i in range(10):
  50.         t = Process(target=task, args=(mutex,))
  51.         t.start()

了解：

分布式锁、悲观锁、乐观锁

九：队列介绍

队列可以并发的派多个线程，对排列的线程处理，并且每个需要处理线程只需要将请求的数据放入队列容器的内存中，线程不需要等待，当排列完毕处理完数据后，线程在准时来取数据即可。

请求数据的线程只与这个队列容器存在关系，处理数据的线程down掉不会影响到请求数据的线程，队列会派给其他线程处理这分数据，它实现了解耦，提高效率。

队列内会有一个有顺序的容器，列表与这个容器是有区别的，列表中数据虽然是排列的，但数据被取走后还会保留，而队列中这个容器的数据被取后将不会保留。

当必须在多个线程之间安全地交换信息时，队列在线程编程中特别有用。

1.作用

解耦：使程序直接实现松耦合，修改一个函数，不会有串联关系。

提高处理效率：ＦＩＦＯ　＝　先进先出，ＬＩＦＯ　＝　后入先出。

2.队列的各个参数

  1. from multiprocessing.queues import Queue
  3. Queue.Queue(maxsize=0)                  # FIFO， 如果maxsize小于1就表示队列长度无限
  5. Queue.LifoQueue(maxsize=0)              # LIFO， 如果maxsize小于1就表示队列长度无限
  7. Queue.qsize()                           # 返回队列的大小
  9. Queue.empty()                           # 如果队列为空，返回True,反之False
  11. Queue.full()                            # 如果队列满了，返回True,反之False
  13. Queue.get([block[, timeout]])           # 读队列，取出数据 ，没有数据将会等待timeout等待时间
  15. Queue.put(item, [block[, timeout]])     # 写队列，放入数据，timeout等待时间
  17. Queue.queue.clear()                     # 清空队列
  19. class queue.PriorityQueue(maxsize=0)    # 存储数据时可设置优先级的队列，优先级设置数越小等级越高
  21. Queue.get(timeout=1)                    # 如果1秒后没取到数据就退出
  23. Queue.get_nowait()                      # 取数据，如果没数据抛queue.Empty异常
  25. Queue.task_done()                       # 后续调用告诉队列，任务的处理是完整的。

3.实例

  1. from multiprocessing import Queue
  3. # 实例化得到要给的对象
  5. q = Queue(5)  # 默认很大，可以放很多，写了个5，只能放5个
  7. # 往管道中放值
  8. q.put(1)
  9. q.put('xxq')
  10. q.put(18)
  11. q.put(18)
  12. q.put(18)
  14. # 从管道中取值
  15. # print(q.get())
  16. # print(q.get())
  17. # print(q.get(timeout=0.1))   # 等0.1s还没有值，就结束
  18. # print(q.get_nowait())       # 不等了，有就是有，没有就是没有
  19. print(q.empty())            # 看一下队列是不是空的    False
  20. print(q.full())             # 看一下队列是不是满的    True

4.总结

  1. q = Queue(队列大小)
  2. # 放值
  3. q.put('xxq')
  4. q.put_nowait(asd)   # 队列满了，放不进去就不放了，报错
  6. # 取值
  7. q.get()     # 从队列头部取出一个值
  8. q.get_nowait()      # 从队列头部取值，没有就报错
  10. # 队列是否为空，是否满
  11. print(q.empty())    # 看一下队列是不是空的
  12. print(q.full())     # 看一下队列是不是满的

5.了解

消息队列（Message Queue）：RabbitMQ、Kafka

十：IPC机制

1.简介

IPC是Inter-Process Communication的缩写，含义为进程间通信或者跨进程通信，是指两个进程之间进行数据交换的过程。

2.实现进程间通信的2种方式

管道：pipe

队列：queue(其实就是pipe+lock)

注意：两者实际上都是内存空间，不要往里面放入大数据，只能放数据量较小的消息

3.IPC所解决的问题

1.当多个任务并发的去修改共享数据，就可能会造成数据错乱,我们通过加互斥锁使多个任务对共享数据的操作由并发变为“串行”，从而保证了共享数据的安全，而当出现需要修改多个共享数据的需求时，我们就得再次加锁处理

——>IPC帮我们解决了需要自己加锁的问题

2.进程间的内存空间是彼此隔离的，如何完成通信(数据交互)，就需要寻求一种共享的东西，硬盘是共享的，但是读取硬盘的速度慢

——>IPC实现了一种内存空间上的共享(两个进程之间通过队列交流)

4.实例

  1. from multiprocessing import Process, Queue
  2. import os
  5. def task1(q):
  6.     print(f'I\'m task1,my id is {os.getpid()}')
  7.     q.put('Hello World')
  10. def task2(q):
  11.     # res=q.get()
  12.     print(f'I\'m task2,my id is {os.getpid()}')
  15. if __name__ == '__main__':
  16.     q = Queue(5)
  18.     t1 = Process(target=task1, args=(q,))
  19.     t1.start()
  20.     t2 = Process(target=task2, args=(q,))
  21.     t2.start()
  23.     print(f'I\'m Main Process,I\'ll print {q.get()}')
  26. # I'm task2,my id is 49516
  27. # I'm task1,my id is 43284
  28. # I'm Main Process,I'll print Hello World

十一：生产者消费者模型

在并发编程中 使用生产者和消费者模式能够解决绝大多数并发问题。

该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

1.什么是生产者消费者模式？

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。

生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

2.为什么要使用生产者和消费者模式？

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。

在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。

同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。

为了解决这个问题，就在生产者和消费者之间通过队列，增加缓冲，避免了生产者和消费者 之间的交互。于是引入了生产者和消费者模式。

Python并发编程-进程 - 图4

3.用队列实现生产者消费者模型

入门版 单生产者 单消费者 - 有BUG

  1. from multiprocessing import Process, Queue
  2. import random
  3. import time
  6. def producer(p_name, p_food, q):
  7.     for i in range(10):
  8.         data = f'[{p_name}]制造了[{p_food}]'
  9.         time.sleep(random.randint(1, 2))  # 模拟制造食物延迟
  10.         print(data)
  11.         q.put(p_food)
  14. def consumer(c_name, q):
  15.     while True:
  16.         c_food = q.get()
  17.         time.sleep(random.randint(1, 2))  # 模拟消费食物延迟
  18.         print(f'[{c_name}]消费了[{c_food}]')
  21. if __name__ == '__main__':
  22.     q = Queue
  24.     p1 = Process(target=producer, args=('xxq', 'Noodles', q))
  25.     p1.start()
  27.     c1 = Process(target=consumer, args=('darker', q))
  28.     c1.start()
  31. # BUG：如果消费者消费完了队列中的食物，队列食物为空，消费者就会停滞

基础版 单生产者 单消费者 - 有BUG

  1. from multiprocessing import Process, Queue
  2. import random
  3. import time
  6. def producer(p_name, p_food, q):
  7.     for i in range(10):
  8.         data = f'[{p_name}]制造了[{p_food}]'
  9.         time.sleep(random.randint(1, 2))  # 模拟制造食物延迟
  10.         print(data)
  11.         q.put(p_food)
  12.     q.put(None)
  15. def consumer(c_name, q):
  16.     while True:
  17.         c_food = q.get()
  18.         if c_food is None:
  19.             return
  20.         time.sleep(random.randint(1, 2))  # 模拟消费食物延迟
  21.         print(f'[{c_name}]消费了[{c_food}]')
  24. if __name__ == '__main__':
  25.     q = Queue()
  27.     p1 = Process(target=producer, args=('xxq', 'Noodles', q))
  28.     p1.start()
  30.     c1 = Process(target=consumer, args=('darker', q))
  31.     c1.start()

进阶版 单生产者 多消费者

  1. from multiprocessing import Process, Queue
  2. import random
  3. import time
  6. def producer(p_name, p_food, q):
  7.     for i in range(10):
  8.         data = f'[{p_name}]制造了[{p_food}]'
  9.         time.sleep(random.randint(1, 2))  # 模拟制造食物延迟
  10.         print(data)
  11.         q.put(p_food)
  14. def consumer(c_name, q):
  15.     while True:
  16.         c_food = q.get()
  17.         if c_food is None:
  18.             return
  19.         time.sleep(random.randint(1, 2))  # 模拟消费食物延迟
  20.         print(f'[{c_name}]消费了[{c_food}]')
  23. if __name__ == '__main__':
  24.     q = Queue()
  26.     p1 = Process(target=producer, args=('xxq', 'Noodles', q))
  27.     p1.start()
  29.     c1 = Process(target=consumer, args=('darker', q))
  30.     c1.start()
  31.     c2 = Process(target=consumer, args=('ben', q))
  32.     c2.start()
  34.     # 生产者生产完毕，放2个None
  35.     p1.join()   # p1进程执行完成，再放None
  36.     q.put(None)
  37.     q.put(None)

高阶版 多生产者 多消费者

  1. from multiprocessing import Process, Queue
  2. import random
  3. import time
  6. def producer(p_name, p_food, q):
  7.     for i in range(10):
  8.         data = f'[{p_name}]制造了[{p_food}]'
  9.         time.sleep(random.randint(1, 2))  # 模拟制造食物延迟
  10.         print(data)
  11.         q.put(p_food)
  14. def consumer(c_name, q):
  15.     while True:
  16.         c_food = q.get()
  17.         if c_food is None:
  18.             return
  19.         time.sleep(random.randint(1, 2))  # 模拟消费食物延迟
  20.         print(f'[{c_name}]消费了[{c_food}]')
  23. if __name__ == '__main__':
  24.     q = Queue()
  26.     p1 = Process(target=producer, args=('xxq', 'Noodles', q))
  27.     p1.start()
  28.     p2 = Process(target=producer, args=('zsy', 'Pizza', q))
  29.     p2.start()
  31.     c1 = Process(target=consumer, args=('darker', q))
  32.     c1.start()
  33.     c2 = Process(target=consumer, args=('ben', q))
  34.     c2.start()
  35.     c3 = Process(target=consumer, args=('Alen', q))
  36.     c3.start()
  38.     # 生产者生产完毕，放2个None
  39.     p1.join()  # p1进程执行完成，再放None
  40.     p2.join()  # p2进程执行完成，再放None
  41.     q.put(None)
  42.     q.put(None)
  43.     q.put(None)

终极版 多生产者 多消费者

  1. """
  2. 生产者:生产/制造东西的
  3. 消费者:消费/处理东西的
  4. 该模型除了上述两个之外还需要一个媒介
  5.     例子：自助餐
  6.     厨师（生产者）制作菜品之后，会将菜品放到餐台（消息队列）
  7.     顾客（消费者）想要吃，可以直接到餐台（消息队列）去拿菜品
  8.     在这个过程中，厨师（生产者）和顾客（消费者）并无直接联系
  9.     而是以餐台（消息队列）为媒介的
  11. 生产者(厨师) + 消息队列(餐台) + 消费者(顾客)
  12. """
  14. from multiprocessing import JoinableQueue, Process
  15. import random
  16. import time
  19. def producer(p_name, p_food, q):
  20.     for i in range(10):
  21.         print(f'{p_name} 生产了 {p_food}')
  22.         # 模拟网络延迟
  23.         time.sleep(random.randint(1, 2))
  24.         # 将数据加入队列
  25.         q.put(p_food)
  28. def consumer(c_name, q):
  29.     while True:
  30.         food = q.get()  # 假如没有数据,程序就会在此处进入堵塞态
  31.         time.sleep(random.randint(1, 2))    # 模拟消费的延迟
  32.         print(f'{c_name} 消费了 {food}')
  33.         q.task_done()  # 告诉队列你已经从里面取出了一个数据并且处理完毕了
  36. if __name__ == '__main__':
  37.     q = JoinableQueue()
  39.     p1 = Process(target=producer, args=('Cook1', 'Beef', q))
  40.     p2 = Process(target=producer, args=('Cook2', 'Cake', q))
  42.     c1 = Process(target=consumer, args=('Alan', q))
  43.     c2 = Process(target=consumer, args=('Ben', q))
  44.     c3 = Process(target=consumer, args=('Cindy', q))
  46.     p1.start()
  47.     p2.start()
  49.     c1.daemon = True    # 将消费者设置成守护进程
  50.     c2.daemon = True
  51.     c3.daemon = True
  52.     c1.start()
  53.     c2.start()
  54.     c3.start()
  56.     p1.join()
  57.     p2.join()
  58.     # 等待生产者生产完毕之后 往队列中添加特定的结束符号
  59.     # q.put(None) # 有多少消费者就往队列中添加多少None,新添加的None必定在队列的末尾
  61.     q.join()  # 等待队列中所有的数据被取完再执行往下执行代码
  62.     """
  63.     JoinableQueue 每当你往该队列中存入数据的时候 内部会有一个计数器+1
  64.     没当你调用task_done的时候 计数器-1
  65.     q.join() 当计数器为0的时候 才往后运行
  66.     """
  67.     # 只要q.join执行完毕 说明消费者已经处理完数据了  消费者就没有存在的必要了

十二：进程池

  1. from concurrent.futures import ProcessPoolExecutor
  2. pool = ProcessPoolExecutor(2)
  3. pool.submit(get_pages, url).add_done_callback(call_back)