python学习并发编程
1、并发编程理论基础
- 并发编程得应用：
- 网络应用：爬虫（直接应用并发编程）
- 网络架构django、flask、tornado 源码-并发编程
- socket server 源码-并发编程
- 计算机操做系统发展史
- 手工操做-读穿孔的纸带、用户独占全机、cpu等待手工操做、cpu利用不充分
- 批处理-磁带存储（联机批处理{一边写入磁带，一边从磁带读出计算}、脱机批处理、只能运行一个程序，遇到IO就等待闲置
- 多道程序系统
- 同时执行多个任务，遇到io就切换，
- 即空间隔离（多个程序运行），时空复用（IO切换）的特色
- 分时系统
- 同时执行多个任务，没有遇到IO也切换，固定时间片到了就切
- 时间片轮转
- 多路性、交互性、独立性、及时性
- 切会浪费cpu时间，下降了cpu效率，但提升了用户体验。
- 实时系统
- 及时响应
- 高可靠性
- 时刻等待响应，即时处理
- 通用操做系统：
- 多道批处理
- 分时
- 实时
- 操做系统分类：
- 我的计算机操做系统
- 网络操做系统
- 分布式操做系统
- 操做系统的做用：
- 是一个协调、管理、和控制计算机硬件资源和软件资源的控制程序。【调度硬件资源、调用管理软件】
- 隐藏了丑陋的硬件调用接口，提供良好的抽象接口
- 管理、调度进程，并将多个进程对硬件的竞争变得有序
- I/O操做：(相对内存来讲得)
- 输入【读到内存中】： input、f.read、accept、recv、connect
- 输出【从内存读出】：print、f.write、send、connect
- 文件操做/网络操做都是IO操做
- 异步：两个任务同时运行（并行）
- 同步：多个任务 串行 【按顺序执行】
- 阻塞：等待 input、accept、recv
- 非阻塞：不等待，直接执行
- 并行：多个任务同时执行【多个CPU在同时执行任务】
- 并发：只有一个CPU，交替执行多个任务【宏观上同时执行，实际是轮转】
- https://www.bughui.com/2017/08/23/difference-between-concurrency-and-parallelism/
- 程序：没有运行
- 进程：运行中得程序、计算机中最小得资源分配单元
- 进程调度：
- 多个进程（运行中得程序）在操做系统得控制下被CPU执行，去享用计算机资源
- 先来先服务调度算法
- 短做业优先
- 时间片轮转
- 多级反馈队列
- 进程调度得过程是不能随意被程序影响得
- 进程三大状态
- 就绪 程序开始运行，进入就绪队列，等待cpu分配时间
- 运行 执行进程，单个时间片内运行完成，就释放资源，没有左右完，就又自动切换到就绪队列等待下次得调度
- 阻塞 执行中得进程遇到事件入IO等致使没法执行，进入阻塞状态，解除后进入就绪队列等待进程调度
- 进程与父进程
- 进程 PID 经过os.getpid() 能够获得
- 父进程 PPID 负责回收一些子进程得资源后才关闭 经过os.getppid()
2、python中的并发编程-进程
- 包：multiprocessing
- 是python中操做、管理进程
- 建立进程、进程同步、进程池、进程之间数据共享
- python中的进程建立
- 对Windows来讲建立子进程，是将主进程内存空间中的全部变量import子进程内存空间中，（import就会执行文件代码）
- windows平台建立子进程必须放在 if __name__ == '__main__' 下执行（否则就会循环建立子进程致使系统崩溃）
- 对Linux来讲建立子进程，就是将主进程内存空间中全部复制过去，复制不会执行内存空间中的代码
- Linux平台就无所谓了，反正只是复制
-
建立进程：异步运行
- multiprocess.Process模块
- Process() 由该类实例化获得得对象，表示一个子进程中得任务（还没有启动）
- p = Process(group，target= func，agrs=（'name',），kwargs={}，name=）
- group 值默认始终为None
- target 表示调用对象
- args 调用对象得位置参数 元组！！！
- kwargs 调用对象得字典
- name为子进程得名称
- 注：必须用关键字方式指定参数、位置参数必须是元组格式，那怕是一个参数也要写成（n,)
- 方法：
- p.start() : 启动进程，调用操做系统的命令，传达要建立进程的申请，并调用子进程中的p.run()方法
- p.run(): 进程启动时运行的方法，其是真正去调用target指定的函数，自定义类，就须要本身定义run方法
- p.terminate(): 强制终止进程p
- p.is_alive(): 查看进程状态值，True表示运行中
- p.join(): 主线程等待子进程p运行结束后才运行 阻塞
- 实操：
-
建立子进程复制代码 1 import os 2 import time 3 from multiprocessing import Process 4 5 6 def f(name): 7 print('in f', os.getpid(), os.getppid()) 8 print('i am is son process', name) 9 10 11 if __name__ == '__main__': 12 p = Process(target=f, args=('小青',)) 13 p.start() # start不是运行一个程序,而是调用操做系统的命令,要建立子进程 14 15 print('我是主程序……') 》》》结果： 我是主程序…… in f 6016 12312 i am is son process 小青 结论：在另外一个地方开辟内存空间，执行f函数，主进程不会阻塞等待
- 给子进程传参：Process（target= func ，args = (参1，））必须是元组结构！！！
-
代码演练
1 def f(args): 2 print('in func 2', args, os.getpid(), os.getppid()) 3 4 5 if __name__ == '__main__': 6 print('我在主进程中……') 7 p1 = Process(target=f, args=(666,)) 8 p1.start() 9 p2 = Process(target=f, args=(777,)) 10 p2.start() 11 print('我会在子进程前运行……，由于我和他们是隔离的，不会等他们') 12 13 ''' 14 结果： 15 我在主进程中…… 16 我会在子进程前运行……，由于我和他们是隔离的，不会等他们 17 in func 2 666 8144 7660 18 in func 2 777 10820 7660 19 '''
- 建立多个子进程：start()方法不是执行方法，而是建立子进程函数，真正执行任务函数是run()方法
-
代码演练
1 def fu(num): 2 print("我是进程 :%d " % num) 3 4 5 if __name__ == '__main__': 6 print('in main', os.getpid(), os.getppid()) 7 for i in range(10): 8 Process(target=fu, args=(i,)).start() 9 print('main 66666') 10 """ 11 结果： 12 in main 11984 2064 13 main 66666 14 我是进程 :1 15 我是进程 :0 16 我是进程 :8 17 我是进程 :7 18 我是进程 :4 19 我是进程 :2 20 我是进程 :9 21 我是进程 :3 22 我是进程 :6 23 我是进程 :5 24 p.start()并无当即执行，而是进入就绪队列，等带cpu调度，因此不是有序的 25 """
- join方法：阻塞、等待子进程执行完毕后再执行后面的代码
-
代码演练
1 def g(args): 2 print("in g", args) 3 4 5 if __name__ == '__main__': 6 print('in main') 7 p = Process(target=g, args=(888,)) 8 p.start() 9 p.join() 10 print('i am in main process') 11 ''' 12 in main 13 in g 888 14 i am in main process 15 结：join老是等子进程执行完毕后再执行接下来的代码 16 '''
- 多进程中运用join方法：必须保证全部子进程结束，而操做系统建立子进程顺序是不定的，故需遍历每一个子进程，每一个都进行join一下
-
代码演练
1 def pro(i): 2 print('in func', i, os.getpid(), os.getppid()) 3 4 5 if __name__ == '__main__': 6 print('in main', os.getpid(),os.getppid()) 7 p_list = [] 8 for i in range(10): 9 p = Process(target=pro, args=(i,)) 10 p.start() # 不是运行一个程序，而是调用操做系统命令，要建立子进程，等待操做系统做业，非阻塞 11 p_list.append(p) 12 print(p_list) 13 for p in p_list: # 遍历每一个子进程，每一个join一下，若是该子进程已经接收，join失效至关于pass，遍历完成就能保证每一个子进程都结束了 14 p.join() # 阻塞，直到p这个子进程执行完毕以后再继续执行 15 print('主进程……') 16 ''' 17 in main 1480 2064 18 [<Process(Process-1, started)>, <Process(Process-2, started)>, <Process(Process-3, started)>, <Process(Process-4, started)>, <Process(Process-5, started)>, <Process(Process-6, started)>, <Process(Process-7, started)>, <Process(Process-8, started)>, <Process(Process-9, started)>, <Process(Process-10, started)>] 19 in func 3 6108 1480 20 in func 7 13756 1480 21 in func 5 12548 1480 22 in func 8 12116 1480 23 in func 4 10948 1480 24 in func 6 11744 1480 25 in func 9 11244 1480 26 in func 1 3968 1480 27 in func 2 9412 1480 28 in func 0 14024 1480 29 主进程…… 30 '''
- p.is_alive() 和 p.terminate() :查看子进程生命状态及强制结束子进程（terminate是非阻塞，并不会等待子进程完全结束才执行后面的代码，只是发一信息要终结，就无论了）
-
代码演练
1 # p.is_alive方法：查看子进程生存状态 2 # p.terminate() 强制结束子进程--非阻塞 3 def gro(i): 4 time.sleep(1) 5 print('in func', i, os.getpid(), os.getppid()) 6 7 8 if __name__ == '__main__': 9 print("in main") 10 p1 = Process(target=gro, args=(1,)) 11 p1.start() 12 # time.sleep(2) # 若是等待一下子，就会执行函数，若是不等，就无论操做系统去建子进程，而直接执行后面的代码，因此可能比建立子进程前就执行了 13 print(p1.is_alive()) # 检测子进程是否还在执行任务 14 p1.terminate() # 强制结束子进程，非阻塞，不会等待状态改变，会立刻执行后面代码 15 print(p1.is_alive()) 16 print('主进程的代码执行结束了……') 17 ''' 18 in main 19 True 20 True 21 主进程的代码执行结束了…… 22 23 结：由于直接执行，主进程执行快些，子进程函数不会执行 24 '''
- 经过面向对象的方法建立子进程（重点：重写run方法，继承Process类，继承父类__init__方法）
-
代码演练
1 class Myprocess(Process): 2 def __init__(self, name): 3 super().__init__(self) # 需继承父类的init方法 4 self.name = name # 添加须要本身的属性 5 6 def run(self): 7 print(self.name) # 只有重写run方法才能将参数传入 8 print(os.getppid(), os.getpid()) 9 10 11 if __name__ == '__main__': 12 p = Myprocess('小强') 13 p.start()
- 进程与进程之间内存中的数据是隔离的！！！
- 进程与进程之间是不能自由的交换内存数据的【内存空间是不能共享的】
- 全局的变量在子进程中修改，其余进程是感知不到的【子进程的执行结果父进程是获取不到的】
- 进程与进程之间想要同行，必须借用其余手段，且两个进程都是自愿的【父子进程通讯是经过socket】
-
代码演练
1 from multiprocessing import Process 2 3 n = 100 4 5 6 def func(): 7 global n 8 n = n - 1 9 return 111 10 11 12 if __name__ == '__main__': 13 n_l = [] 14 for i in range(100): 15 p = Process(target=func) 16 p.start() 17 n_l.append(p) 18 for p in n_l: p.join() 19 print(n) 20 21 结果为：100 22 23 总结：说明子进程没法改变主进程的全局变量，本质是没法自由通讯，但子进程中的n确定减小了，只是无法拿出来
- 守护进程 p.daemon = True
- 特色：守护进程的生命周期只和主进程的代码有关系，和其余子进程没有关系，会随着主进程结束而结束
- 做用：报活，监控主进程生命状态
- 主进程建立守护进程
- 守护进程会在主进程代码结束后就终止
- 守护进程内没法再开子进程，否咋抛出异常 AssertionError: daemonic processes are not allowed to have children
- 守护进程的属性，默认是False，若是设置程True，就表示设置这个子进程为一个守护进程
- 设置守护进程的操做应该在开子进程以前即p.start()以前
-
代码演练
1 from multiprocessing import Process 2 import time 3 def func1(): 4 print('begin') 5 time.sleep(3) 6 print('wawww') 7 8 # if __name__ == '__main__': 9 # p = Process(target=func1) 10 # # p.daemon = True 11 # p.start() 12 # time.sleep(1) 13 # print('in main') 14 ''' 15 结果： 16 begin 17 in main 18 19 结论：守护进程随着主进程结束而结束，那怕守护进程任务没有执行完毕 20 ''' 21 22 def f1(): 23 print('begin fun1') 24 time.sleep(3) 25 print('baidu') 26 27 def f2(): 28 while True: 29 print('in f2') 30 time.sleep(0.5) 31 32 if __name__ == '__main__': 33 Process(target=f1,).start() 34 p = Process(target=f2) 35 p.daemon = True 36 # 守护进程的属性,默认是False,若是设置成True,就表示设置这个子进程为一个守护进程 37 # 设置守护进程的操做应该在开启子进程以前 38 p.start() 39 time.sleep(1) 40 print('in main') # 主进程in main执行完后，守护进程就会结束，但主进程并无结束而是等另外一个子进程结束后才结束 41 42 43 # 设置成守护进程以后 会有什么效果呢? 44 # 守护进程会在主进程的代码执行完毕以后直接结束,不管守护进程是否执行完毕 45 46 # 应用 47 # 报活 主进程还活着 48 # 100台机器 100个进程 10000进程 49 # 应用是否在正常工做 - 任务管理器来查看 50 # 守护进程如何向监测机制报活???send/写数据库 51 # 为何要用守护进程来报活呢?为何不用主进程来工做呢??? 52 # 守护进程报活几乎不占用CPU,也不须要操做系统去调度 53 # 主进程不能严格的每60s就发送一条信息
-
进程同步控制
- 进程的同步控制 - 进程之间有一些简单的信号传递,可是用户不能感知,且用户不能传递本身想传递的内容
- 锁：multiprocessing.Lock ********* 【互斥锁】
- lock = Lock() # 创造一把锁
- lock.acquire() # 获取了这把锁的钥匙、阻塞，锁未换就一直阻塞着
- lock.release() # 归还这把钥匙
- 解决多个进程共享一段数据的时候，数据会出现不安全的现象，经过加锁来维护数据的安全性，同一刻，只容许一个线程修改数据
-
代码演练-抢票的故事
1 import json 2 import time 3 from multiprocessing import Lock 4 from multiprocessing import Process 5 6 # 锁 7 lock = Lock() # 创造了一把锁 8 lock.acquire() # 获取了这把锁的钥匙 9 lock.release() # 归还这把钥匙，其余进程就能够拿锁了 10 11 12 # 抢票的故事 13 # 需求：每一个人都能查看余票、买相同车次票同一刻只能一人买完，另外一人才能买 14 15 16 def search(i): 17 with open('db', encoding='utf-8') as f: 18 count_dic = json.load(f) 19 time.sleep(0.2) # 模拟网络延迟 20 print('person %s 余票：%s 张' % (i, count_dic.get('count'))) 21 return count_dic.get('count'), count_dic # 返回余票数，及字典 22 23 24 def buy(i): 25 count, count_dict = search(i) 26 if count > 0: 27 count_dict['count'] -= 1 # 有票就能够买 28 print('person %s 买票成功'% i) 29 time.sleep(2) 30 with open('db', 'w', encoding='utf-8') as f: 31 json.dump(count_dict, f) # 更改余票额度 32 33 34 def task(i, lock): 35 search(i) 36 lock.acquire() # 若是以前已经被acquire了 且 没有被release 那么进程会在这里阻塞 37 buy(i) 38 lock.release() 39 40 41 if __name__ == '__main__': 42 lock = Lock() 43 for i in range(1, 11): 44 Process(target=task, args=(i, lock)).start() 45 46 # 当多个进程共享一段数据的时候,数据会出现不安全的现象, 47 # 须要加锁来维护数据的安全性 48 49 ''' 50 D:\install\Python36\python.exe D:/install/project/七、并发编程/三、锁.py 51 person 6 余票：5 张 52 person 5 余票：5 张 53 person 8 余票：5 张 54 person 2 余票：5 张 55 person 4 余票：5 张 56 person 10 余票：5 张 57 person 1 余票：5 张 58 person 9 余票：5 张 59 person 3 余票：5 张 60 person 7 余票：5 张 61 person 6 余票：5 张 62 person 6 买票成功 63 person 5 余票：4 张 64 person 5 买票成功 65 person 8 余票：3 张 66 person 8 买票成功 67 person 2 余票：2 张 68 person 2 买票成功 69 person 4 余票：1 张 70 person 4 买票成功 71 person 10 余票：0 张 72 person 1 余票：0 张 73 person 9 余票：0 张 74 person 3 余票：0 张 75 person 7 余票：0 张 76 77 Process finished with exit code 0 78 '''
注：进程间的数据交互，本质也用到了socket通讯，不过都是本地的，基于文件的，能够经过将py名写成socket来看报错得知。node
-
1 #加锁能够保证多个进程修改同一块数据时，同一时间只能有一个任务能够进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。 2 虽然能够用文件共享数据实现进程间通讯，但问题是： 3 1.效率低（共享数据基于文件，而文件是硬盘上的数据） 4 2.须要本身加锁处理 5 6 #所以咱们最好找寻一种解决方案可以兼顾：一、效率高（多个进程共享一块内存的数据）二、帮咱们处理好锁问题。这就是mutiprocessing模块为咱们提供的基于消息的IPC通讯机制：队列和管道。 7 队列和管道都是将数据存放于内存中 8 队列又是基于（管道+锁）实现的，可让咱们从复杂的锁问题中解脱出来， 9 咱们应该尽可能避免使用共享数据，尽量使用消息传递和队列，避免处理复杂的同步和锁问题，并且在进程数目增多时，每每能够得到更好的可获展性。 - 信号量：multiprocessing.Semaphore
- 资源有限，同时容许必定数量的进程访问修改数据，即一把锁对应多把钥匙
- 本质：锁+ 计数器
-
代码演练-kvt的故事
1 互斥锁同时只容许一个线程更改数据，而信号量Semaphore是同时容许必定数量的线程更改数据 。 2 假设商场里有4个迷你唱吧，因此同时能够进去4我的，若是来了第五我的就要在外面等待，等到有人出来才能再进去玩。 3 实现： 4 信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。这是迪科斯彻（Dijkstra）信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器这样的有限资源。 5 信号量与进程池的概念很像，可是要区分开，信号量涉及到加锁的概念 6 7 8 import time, random 9 from multiprocessing import Semaphore, Process 10 11 # ktv 只有4个房间，即同时只能四我的进去，其余人必须等其中的人出来才能进去 12 # 13 # sem = Semaphore(4) # 设置信号量个数，并发数 14 # sem.acquire() 15 # print('进去1我的，关门阻塞中') 16 # sem.acquire() 17 # print('进去第2我的，关门阻塞中') 18 # sem.acquire() 19 # print('进去第3我的，关门阻塞中') 20 # sem.acquire() 21 # print('进去第4我的，关门阻塞中') 22 # sem.release() # 必须归还一把，才能继续下面的代码，否则一直阻塞中 23 # sem.acquire() 24 # print(6666) 25 # sem.release() 26 ''' 27 D:\install\Python36\python.exe D:/install/project/七、并发编程/四、信号量.py 28 进去1我的，关门阻塞中 29 进去第2我的，关门阻塞中 30 进去第3我的，关门阻塞中 31 进去第4我的，关门阻塞中 32 6666 33 34 Process finished with exit code 0 35 ''' 36 37 38 def ktv(num, sem): 39 sem.acquire() 40 print('person %s 进入了ktv' % num) 41 time.sleep(random.randint(1, 4)) 42 print('person %s 进出了ktv' % num) 43 sem.release() 44 45 46 if __name__ == '__main__': 47 sem = Semaphore(4) 48 for i in range(10): 49 Process(target=ktv, args=(i, sem)).start() 50 51 ''' 52 最开始是4个同时进入，以后又人出，才能有人进 53 D:\install\Python36\python.exe D:/install/project/七、并发编程/四、信号量.py 54 person 2 进入了ktv 55 person 8 进入了ktv 56 person 9 进入了ktv 57 person 7 进入了ktv 58 person 2 进出了ktv 59 person 6 进入了ktv 60 person 7 进出了ktv 61 person 5 进入了ktv 62 person 8 进出了ktv 63 person 1 进入了ktv 64 person 9 进出了ktv 65 person 0 进入了ktv 66 person 1 进出了ktv 67 person 4 进入了ktv 68 person 6 进出了ktv 69 person 3 进入了ktv 70 person 0 进出了ktv 71 person 5 进出了ktv 72 person 4 进出了ktv 73 person 3 进出了ktv 74 75 Process finished with exit code 0 76 '''
-
事件：multiprocessing.Eventpython
-
定义：全局定义了一个‘flag’，若是标志为False，当程序执行event.wait方法时就会阻塞，若是为True，那么event.wait方法时便再也不阻塞。redis
- 做用：主进程控制其余线程的执行，实现两个或多个线程间的交互
- 使用：
- 事件创立之初，默认是Faslse，即阻塞状态
- e = Event() # 建立事件，默认False
- print(e.is_set()) # 查看状态
- e.set() # 将标志设置为True
- e.clear() # 将标志设置为False
- e.wait() # 等待，当标志为False，那么阻塞，当标志为True，那么非阻塞，wait什么也不作直接pass
- e.wait(timeout=10) # 若是信号在阻塞10s以内变为True，那么就不继续阻塞直接pass，若是就阻塞10s以后状态仍是没变，那么继续阻塞
-
代码演练-红绿灯的故事
1 from multiprocessing import Process, Event 2 import time, random 3 4 5 def car(e, n): 6 while True: 7 if not e.is_set(): # 进程刚开启，is_set()的值是Flase，模拟信号灯为红色 8 print('\033[31m红灯亮\033[0m，car%s等着' % n) 9 e.wait() # 阻塞，等待is_set()的值变成True，模拟信号灯为绿色 10 print('\033[32m车%s 看见绿灯亮了\033[0m' % n) 11 time.sleep(random.randint(3, 6)) 12 if not e.is_set(): #若是is_set()的值是Flase，也就是红灯，仍然回到while语句开始 13 continue 14 print('车开远了,car', n) 15 break 16 17 18 def police_car(e, n): 19 while True: 20 if not e.is_set():# 进程刚开启，is_set()的值是Flase，模拟信号灯为红色 21 print('\033[31m红灯亮\033[0m，car%s等着' % n) 22 e.wait(0.1) # 阻塞，等待设置等待时间，等待0.1s以后没有等到绿灯就闯红灯走了 23 if not e.is_set(): 24 print('\033[33m红灯,警车先走\033[0m，car %s' % n) 25 else: 26 print('\033[33;46m绿灯，警车走\033[0m，car %s' % n) 27 break 28 29 30 31 def traffic_lights(e, inverval): 32 while True: 33 time.sleep(inverval) 34 if e.is_set(): 35 print('######', e.is_set()) 36 e.clear() # ---->将is_set()的值设置为False 37 else: 38 e.set() # ---->将is_set()的值设置为True 39 print('***********',e.is_set()) 40 41 42 if __name__ == '__main__': 43 e = Event() 44 for i in range(10): 45 p=Process(target=car,args=(e,i,)) # 建立是个进程控制10辆车 46 p.start() 47 48 for i in range(5): 49 p = Process(target=police_car, args=(e, i,)) # 建立5个进程控制5辆警车 50 p.start() 51 t = Process(target=traffic_lights, args=(e, 10)) # 建立一个进程控制红绿灯 52 t.start() 53 54 print('============》')
-
进程间通讯-队列和管道
- IPC 进程之间的通讯 multiprocessing.Queue/Pipe：进程之间的内存是不共享的，隔离的，但队列、管道有使之通讯的功能。
- 队列 multiprocessing.Queue
-
数据类型中的队列及队列中的方法，在进程、线程中通用
from queue import Queue # 队列 先进先出FIFO,有序 # 应用：维护秩序的时候用的比较多，买票，抢票 q = Queue(5) # 设置队列的长度，即元素个数，即只能放入5个元素 ret = q.qsize() # 得到当前队列中的元素个数，此方法不许，在多进程中，此刻获取结果时，也许其余进程向里面加入了元素 q.put(1111) # 向队列中放入对象，若是队列已满，则阻塞等待，一直到空间可用为止 ''' 参数： item：项目、元素、对象 block：默认True，队列满一直等待阻塞，False则为不阻塞，满则直接主动自定义报错 timeout:阻塞等待的时间，时间到了，还不能放，则报错Queue.Empty异常 ''' q.put_nowait(2222) # 放入元素，满了，不等直接报错 q.get() # 返回q即队列中的元素，队列中为空，则阻塞一直等待有值为止，通向可设置timeout q.get_nowait() # 队列为空时，直接报错 q.empty() # 判断是否为空，空则返回True，一样在多进程中不许 q.full() # 判断是否满了，满了则返回True，多进程中不许，主要是进程是异步操做
- 定义：先进先出、有序
- 本质：管道 + 锁
- 做用：因为先进先出的特色+进程通讯的功能+数据进程安全,常常用它来完成进程之间的通讯
-
代码演练-主子通讯/子子通讯
# 例子1 一进程放 一进程取 from multiprocessing import Queue, Process def con(q): print(q.get()) # 从队列中拿，没有直到等到有，因此那么它比其余进程快，最后也能拿到数据 def pro(q): q.put(112) # 向队列中放入112 if __name__ == '__main__': q = Queue() p = Process(target=con, args=(q,)) p.start() p = Process(target=pro, args=(q,)) p.start() print('我在主进程中……') ''' 看出队列能够实现进程间的通讯 ''' # 主放， 子取 from multiprocessing import Queue, Process def f(q): print(q.get()) if __name__ == '__main__': q = Queue() Process(target=f, args=(q,)).start() # create son_process q.put(666) '''看出主进程可和子进程通讯'''
- 生成者消费者模型：可解决大部分并发问题
- 并发中的问题：
- 生产数据快，消费数据慢，内存空间的浪费，【产生的数据不能丢只能放在内存中等着处理】
- 生产数据慢，消费数据快，效率低下 【老是要等着生产数据，啥也干不了】
- 做用：- 解决创造（生成）数据和处理（消费）数据的效率不平衡问题
- 实现：将创造数据和处理数据放在不一样的进程中，根据他们的效率来调整进程的个数
-
代码演练-生产者生产完告终束，而消费者一直在get阻塞中，待解决
1 import time 2 import random 3 from multiprocessing import Process, Queue 4 5 6 def consumer(q,name): 7 while 1: 8 food = q.get() 9 print('%s 吃了 %s ' % (name, food)) 10 time.sleep(random.random()) 11 12 13 def producer(q,name,food,n=10): 14 for i in range(1, n): # 定义生产10个食物 15 time.sleep(random.random()) # 模拟生产慢，消费快 16 fd = food + str(i) 17 print('%s 生产了 %s' %(name,fd)) 18 q.put(fd) 19 20 if __name__ == '__main__': 21 q = Queue(10) 22 for person in range(6): # 定义消费者多 23 Process(target=consumer, args=(q, 'person'+ str(person))).start() 24 Process(target=producer, args=(q,'小强','米饭')).start() 25 Process(target=producer, args=(q,'小东','面条')).start() 26 Process(target=producer, args=(q,'小hua','面条')).start() 27 Process(target=producer, args=(q,'小cai','面条')).start()
- 问题：
消费者一直在等着拿数据，生产者生产完了就结束了，生产者须要告诉消费者生产完了才合理，即向队列中放入stop信号
- 重点：消费者有多少个，就必需要发多少个stop信号
- 解决：
- 方案一：生产者子进程中发，缺陷：但生产者数必须和消费者数同样，消费者进程才能所有关闭
-
代码演练
1 def consumer(q,name): 2 while 1: 3 food = q.get() 4 if food == 'stop':break 5 print('%s 吃了 %s ' % (name, food)) 6 time.sleep(random.random()) 7 8 9 def producer(q,name,food,n=10): 10 for i in range(1, n): # 定义生产10个食物 11 time.sleep(random.random()) # 模拟生产慢，消费快 12 fd = food + str(i) 13 print('%s 生产了 %s' %(name,fd)) 14 q.put(fd) 15 q.put('stop') 16 17 if __name__ == '__main__': 18 q = Queue(10) 19 for person in range(4): # 定义消费者多 20 Process(target=consumer, args=(q, 'person'+ str(person))).start() 21 Process(target=producer, args=(q,'小强','米饭')).start() 22 Process(target=producer, args=(q,'小东','面条')).start() 23 Process(target=producer, args=(q,'小hua','面条')).start() 24 Process(target=producer, args=(q,'小cai','面条')).start() 25 26 # 生产者加入结束信号，消费者收到后就结束，不存在还有进程在使用数据，由于队列是先进先出的 27 28 # 且有多少个消费者就须要发多少个stop信号，否则就会致使有的的进程还在等待中
- 方案二：主进程中发，但必需要等生产者都结束，那么就加入p.join() ，仍是得有多少个消费者，发多少个stop信号，没法自动
-
代码演练
1 from multiprocessing import Queue, Process 2 import random, time 3 4 5 def consumer(q, name): 6 while 1: 7 food = q.get() 8 if food == 'stop': break 9 print('%s 吃了 %s ' % (name, food)) 10 time.sleep(random.random()) 11 12 13 def producer(q, name, food, n=10): 14 for i in range(1, n): # 定义生产10个食物 15 time.sleep(random.random()) # 模拟生产慢，消费快 16 fd = food + str(i) 17 print('%s 生产了 %s' % (name, fd)) 18 q.put(fd) 19 20 21 if __name__ == '__main__': 22 q = Queue(10) 23 for person in range(4): # 定义消费者4个 24 Process(target=consumer, args=(q, 'person'+ str(person))).start() 25 p1 = Process(target=producer, args=(q,'小强','米饭')) 26 p1.start() 27 p2 = Process(target=producer, args=(q,'小东','面条')) 28 p2.start() 29 p1.join() # 保证p生产者结束 30 p2.join() 31 q.put('stop') # 必须得发四个stop信号 32 q.put('stop') 33 q.put('stop') 34 q.put('stop')
- 方案三：
- JoinableQueue([maxsize])
建立可链接的共享进程队列。这就像是一个Queue对象，但队列容许项目的使用者通知生产者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。 -
JoinableQueue使用方法
1 JoinableQueue的实例p除了与Queue对象相同的方法以外，还具备如下方法： 2 3 q.task_done() 4 使用者使用此方法发出信号，表示q.get()返回的项目已经被处理。若是调用此方法的次数大于从队列中删除的项目数量，将引起ValueError异常。 5 6 q.join() 7 生产者将使用此方法进行阻塞，直到队列中全部项目均被处理。阻塞将持续到为队列中的每一个项目均调用q.task_done()方法为止。 8 下面的例子说明如何创建永远运行的进程，使用和处理队列上的项目。生产者将项目放入队列，并等待它们被处理。
代码演练1 from multiprocessing import Process,JoinableQueue 2 import time,random,os 3 def consumer(q): 4 while True: 5 res=q.get() 6 time.sleep(random.randint(1,3)) 7 print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res)) 8 q.task_done() #向q.join()发送一次信号,证实一个数据已经被取走了 9 10 def producer(name,q): 11 for i in range(10): 12 time.sleep(random.randint(1,3)) 13 res='%s%s' %(name,i) 14 q.put(res) 15 print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res)) 16 q.join() #生产完毕，使用此方法进行阻塞，直到队列中全部项目均被处理。 17 18 19 if __name__ == '__main__': 20 q=JoinableQueue() 21 #生产者们:即厨师们 22 p1=Process(target=producer,args=('包子',q)) 23 p2=Process(target=producer,args=('骨头',q)) 24 p3=Process(target=producer,args=('泔水',q)) 25 26 #消费者们:即吃货们 27 c1=Process(target=consumer,args=(q,)) 28 c2=Process(target=consumer,args=(q,)) 29 c1.daemon=True 30 c2.daemon=True 31 32 #开始 33 p_l=[p1,p2,p3,c1,c2] 34 for p in p_l: 35 p.start() 36 37 p1.join() 38 p2.join() 39 p3.join() 40 print('主') 41 42 #主进程等--->p1,p2,p3等---->c1,c2 43 #p1,p2,p3结束了,证实c1,c2确定全都收完了p1,p2,p3发到队列的数据 44 #于是c1,c2也没有存在的价值了,不须要继续阻塞在进程中影响主进程了。应该随着主进程的结束而结束,因此设置成守护进程就能够了。
- 消费者每消费处理完一个数据，就向生产者发送消息，经过q.task_done
- 生产者每生产一个数据就在计数器+1，每接到消费者发得q.task_done,就-1，消费者消费完了，生产者计数也为0了，故才会结束，经过q.join()
- 为了让做业完成，程序能关闭，等生产者结束，主进程也跟着关闭便可，即生成者.join()一下便可
- 那么就能够把消费者设置成守护进程，主进程结束，守护进程也就结束了，这样就ok了。
- ==》消费者消费完毕
- ==》生产者结束
- ==》主进程代码结束，从而守护进程进程结束
- ==》消费者结束
- 注：q.join() 只是保证生产者结束便可，那么在子进程join和主进程中join就同样了。
-
注：栈 ==> 先进后出----算法算法
-
管道：multiprocessing.Pipe
数据库 - 定义：IPC通讯的一种机制，队列就是基于管道来完成的通讯的，可是管道是原生的通讯方式，在进程之间会产生数据不安全的状况，须要本身手动加锁来处理，管道在数据传输过程当中，还涉及到一个端口管理，这个须要咱们在代码中作处理才能使代码更完善。
- 管道使用中待解决的问题：
- 一、多进程通讯数据不安全
-
答：不安全是因多个进程可能会在同一刻同时去取同一个数据，也可能同一刻拿走一个数据，位置就空了，别的应该放在空位置上，
而因同一刻时，另外一个进程会认为该位置上有数据，就放在后面了，这就会致使数据异常，解决方法就是在拿数据前加锁，拿完归还锁。django - 二、端口问题，生产者或消费者没有使用的端口须要关闭，否则，消费者就会认为还有数据要接收，就会一直recv中，所以就须要生成者关闭数据的输出端，对应的消费者就须要关闭数据的输入端，才能保证当消费者recv不到数据了经过报错EOFerror而中止。
-
代码演练-相关问题注意
1 from multiprocessing import Pipe, Process 2 3 4 # 管道 5 # 队列是基于管道实现的 6 # 队列 进程间数据安全的 7 # 管道 进程间数据不安全的 8 # 队列 = 管道 + 锁 9 10 # left, right = Pipe() 11 # print(right.recv()) 12 # (<multiprocessing.connection.PipeConnection object at 0x000002817A7FB128>, <multiprocessing.connection.PipeConnection object at 0x000002817A65C128>) 13 # 管道对象返回的是一个元组 14 15 16 # 全部的IPC通讯都是经过socket实现的 17 18 # 左边放，右边出，一样能够左收，右发，全双工模式 19 20 # 管道必须在建立进程前建立 21 def consumer(left, right): 22 left.close() # 消费者用右边接那么就把左边关闭 23 while 1: 24 try: 25 print(right.recv()) 26 except EOFError: # 再也接不到数据了，从而报错，才能退出 27 break 28 29 30 if __name__ == '__main__': 31 left, right = Pipe() 32 p = Process(target=consumer, args=(left, right)) 33 p.start() 34 right.close() # 用右边发送，那么左边就关闭 35 for i in range(1, 11): 36 left.send(3333) 37 left.close() # 不用了就关闭 38 39 # EOF异常的触发 40 # 在这一个进程中 若是不在用这个端点了,应该close 41 # 这一在recv的时候,若是其余端点都被关闭了,就可以知道不会在有新的消息传进来 42 # 此时就不会在这里阻塞等待,而是抛出一个EOFError 43 # * close并非关闭了整个管道,而是修改了操做系统对管道端点的引用计数的处理
-
进程间数据共享
-
进程之间数据共享
消息传递的并发是趋势
线程是经过线程集合，用消息队列来交换数据
进程间应尽可能避免通讯，由于可能不安全，想安全就必须加锁，加锁就会影响效率。
redis分布式、数据库解决进程之间数据共享问题
'''
进程间数据是独立的，能够借助于队列或管道实现通讯，两者都是基于消息传递的
虽然进程间数据独立，但能够经过Manager实现数据共享，事实上Manager的功能远不止于此
'''
# Manager模块
# 全部的数据类型 都可以进行数据共享
# 一部分都是不加锁 不支持数据进程安全
# 不安全的解决办法 加锁 -
代码演练
1 from multiprocessing import Manager,Process,Lock 2 def work(d,lock): 3 with lock: 4 d['count'] -= 1 5 6 if __name__ == '__main__': 7 lock = Lock() 8 m = Manager() 9 dic = m.dict({'count':100}) 10 p_l = [] 11 for i in range(100): # 开了100进程 12 p = Process(target=work, args=(dic, lock)) 13 p_l.append(p) 14 p.start() 15 for p in p_l: 16 p.join() 17 print(dic) 18 ''' 19 结果：{'count':0} 20 ''' 21 22 # with as 的机制 23 # __enter__ 24 # __exit__
-
进程池：multiprocessing.Pool
- 概念：
- 几个CPU就能同时运行几个进程，并行， 进程的个数不是无限开启的 会给操做系统调度增长负担【开启进程慢】
- 且真正能被同时执行的进程最多也就和CPU个数相同等
- 进程的开启和销毁都要消耗资源和时间
- 进程池中的进程不会重复开启和关闭，而是一直在那被使用，减小时间及操做系统调度的开销
- 进程池中的进程数保持一致，保证同一时间最多有固定数量的进程再运行，减小操做系统调度难度的同时实现并发
- 若是任务数大于池子中进程数，就只能等着，有点像信号量，但信号量进程是新建的，而进程池中的进程一直在
- 使用场景：
- 面向高计算型的场景，没有或比较少IO型的程序 采用多进程
- 但愿并行 最充分的使用CPU
- 对象方法：
- p = Pool(num) 建立进程池，指定池中进程数
- p.apply(func,args=(i,) ) 同步，会等func返回值后，阻塞，下一个任务才能继续
- p.apply_async(func,args=(j,)) 异步
- p.close() 不是关闭进程池中的进程让进程不工做了，而是关闭进程池，让任务再也不提交，已经接到的任务继续执行直至执行结束
- p.join() 等待进程任务中的全部任务都执行完毕
- 建立一个进程池
-
代码演练
1 import os 2 import time 3 from multiprocessing import Pool 4 print(os.cpu_count()) # 获取cpu个数 5 6 7 def wahaha(): 8 time.sleep(1) 9 print(os.getpid()) 10 return True 11 12 13 if __name__ == '__main__': 14 p = Pool(5) # 进程池中进程数通常为cpu个数或者cpu+1，不要超过10个 15 for i in range(20): 16 # p.apply(func=wahaha) # 同步，通常不用，还不如一个进程去循环作，进程池有返回值，基于ipc通讯，本身能够经过q来通讯 17 p.apply_async(func=wahaha) # async 异步的 18 p.close() # 关闭进程池，进程池中的进程不工做了，让任务不能再继续提交了， 19 p.join() # 等待这个池中提交的任务都执行完，就结束
-
异步提交,不获取返回值编程
-
代码演练
def wahaha(): time.sleep(1) print(os.getpid()) if __name__ == '__main__': p = Pool(5) # CPU的个数 或者 +1 ret_l = [] for i in range(20): ret = p.apply_async(func = wahaha) # async 异步的 ret_l.append(ret) p.close() # 关闭 并非进程池中的进程不工做了 # 而是关闭了进程池,让任务不能再继续提交了 p.join() # 等待这个池中提交的任务都执行完 # # 表示等待全部子进程中的代码都执行完 主进程才结束
-
异步提交，得到返回值，等待全部任务都执行完毕以后再统一获取结果json
-
代码演练
1 # 异步提交,获取返回值,等待全部任务都执行完毕以后再统一获取结果 2 def wahaha(): 3 time.sleep(1) 4 print(os.getpid()) 5 return True 6 7 if __name__ == '__main__': 8 p = Pool(5) # CPU的个数 或者 +1 9 ret_l = [] 10 for i in range(20): 11 ret = p.apply_async(func = wahaha) # async 异步的 12 ret_l.append(ret) 13 p.close() # 关闭 不是进程池中的进程不工做了 14 # 而是关闭了进程池,让任务不能再继续提交了 15 p.join() # 等待这个池中提交的任务都执行完 16 for ret in ret_l: 17 print(ret.get())
-
异步提交，得到返回值，一个任务执行完毕以后就能够获取到一个结果(顺序是按照提交任务的顺序)【用在任务关联不大的时候】flask
-
代码演练，不用p.close()和p.join(),ret.get()会阻塞主进程按顺序一个个获取结果
1 def wahaha(): 2 time.sleep(1) 3 print(os.getpid()) 4 return True 5 6 if __name__ == '__main__': 7 p = Pool(5) # CPU的个数 或者 +1 8 ret_l = [] 9 for i in range(20): 10 ret = p.apply_async(func = wahaha) # async 异步的 11 ret_l.append(ret) 12 for ret in ret_l: 13 print(ret.get())
-
总结bootstrap
-
总结
1 # 异步的 apply_async 2 # 1.若是是异步的提交任务,那么任务提交以后进程池和主进程也异步了, 3 #主进程不会自动等待进程池中的任务执行完毕 4 # 2.若是须要主进程等待,须要p.join 5 # 可是join的行为是依赖close 6 # 3.若是这个函数是有返回值的 7 # 也能够经过ret.get()来获取返回值 8 # 可是若是一边提交一遍获取返回值会让程序变成同步的 9 # 因此要想保留异步的效果,应该讲返回对象保存在列表里,全部任务提交完成以后再来取结果 10 # 这种方式也能够去掉join,来完成主进程的阻塞等待池中的任务执行完毕
-
进程池解决原生socket，同一时刻只能和一个客户端链接【这种方式的弊端是同时最多只能和进程池中的数量相同，其它用户等待】
-
server
1 import socket 2 from multiprocessing import Pool 3 4 5 def talk(conn): 6 try: 7 while 1: 8 msg = conn.recv(1024).decode('utf-8') 9 print(msg) 10 conn.send(b'hello') 11 finally: 12 conn.close() 13 14 15 if __name__ == '__main__': 16 sk = socket.socket() 17 sk.bind(('127.0.0.1', 9999)) 18 sk.listen() 19 pool = Pool(5) 20 try: 21 while 1: 22 conn, addr = sk.accept() 23 pool.apply_async(talk, args=(conn,)) 24 finally: 25 conn.close() 26 sk.close()
-
client
1 import socket 2 3 ip_port = ('127.0.0.1', 9999) 4 sk = socket.socket() 5 sk.connect(ip_port) 6 7 while 1: 8 msg = input('>>>>:').strip() 9 if len(msg) == 0: continue 10 sk.send(msg.encode('utf-8')) 11 content = sk.recv(1024).decode('utf-8') 12 print(content)
-
Pool中的回调函数callback
- 定义：将一个进程的的执行结果的返回值，会当callback参数来执行callback函数，从而减小了ret.get()等I/O操做浪费的时间了。
- 做用：进程池中任何一个任务一旦处理完了，就当即告知主进程，我已执行完毕，你能够处理个人结果了。主进程则调用一个函数去处理该结果，该函数就是回掉函数。
- 重点：回调函数是主进程中执行【若是有两个任务,个人第二个任务在第一个任务执行完毕以后可以当即被主进程执行】
-
代码演练
1 import os 2 import time 3 import random 4 from multiprocessing import Pool 5 6 # 异步提交,获取返回值,从头至尾一个任务执行完毕以后就能够获取到一个结果 7 def wahaha(num): 8 time.sleep(random.random()) 9 print('pid : ',os.getpid(),num) 10 return num 11 12 def back(arg): 13 print('call_back : ',os.getpid(),arg) 14 15 if __name__ == '__main__': 16 print('主进程',os.getpid()) 17 p = Pool(5) # CPU的个数 或者 +1 18 for i in range(20): 19 ret = p.apply_async(func = wahaha,args=(i,),callback=back) # async 异步的 20 p.close() 21 p.join() 22 23 # 回调函数 _ 在主进程中执行 24 # 在发起任务的时候 指定callback参数 25 # 在每一个进程执行完apply_async任务以后,返回值会直接做为参数传递给callback的函数,执行callback函数中的代码 26 27 28 # 北京 30min 30min +5min 5min + 5min 29 # 建设 20min 直接办 + 5min 5min 30 # 中国 1h 20min +5 25min + 5min 31 # 农业 2h 55min + 5min 55min + 5min 32 # 工商 15min 5min 15min + 5min 33 34 # 2h10min 35 # 2h05min
3、python中的并发编程-线程
-
程序、进程、线程
1 程序：是指令的集合，它是进程运行的静态描述文本 2 进程：是程序的一次执行活动，是动态概念，计算机中最小资源分配单位 3 线程：cpu调度的最小单位，从属于进程，任务的实际执行者
- 进程：计算机中最小的资源分配单位，在利用多个cpu执行的过程当中，对多个程序的资源进行管理和隔离
- 优势：实现并发，多进程异步执行多个任务，内存地址隔离，IPC通讯经过Queue，Pipe
- 缺陷：创立、撤销、切换都会有较大的时空开销，过多的进程会形成操做系统调度的压力，同一时刻，只能运行一个任务
- 线程：cpu 调度的最小单位
- 特色：一、轻型实体（包含程序、数据、tcb，基本不占资源）
- 二、独立调度和分派的基本单位（线程与线程的切换很是快，开销小）
- 三、共享进程资源，同一进程内线程间可相互通讯
- 四、并发执行
- 进程与线程区别：
- 一、线程属于进程、进程负责获取操做兄台那个分配的资源、线程负责任务执行
- 二、每一个进程中至少有一个线程
- 三、地址空间和其余资源，多进程之间内存相互隔离，多线程间共享同一进程数据
- 四、通讯，进程间不能自由通讯，通讯需借助管道，队列等工具，线程能够直接读写，自由通讯，共用主进程Id
- 五、调度和切换，多进程开启、结束时间开销大、切换效率低，多线程开销小、切换效率高
- 六、多线程操做系统中，进程不是一个可执行的实体
- cpython 解释器下有全局解释器锁
- 在同一个进程中的多个线程在同一时刻只能有一个线程访问cpu，致使多线程没法并行处例任务，利用多核处理器
- 注：程序计算的时候才会用到cpu，sleep或I/O阻塞的时候是不会用到CPU的
- jpython/pypy解释器没有全局解释器锁
-
python线程管理：threading模块
- 建立线程：共享主进程资源，故每一个线程的进程ID同样，但各个线程有本身的id经过get_ident()，需导入类get_ident
-
代码演练-方式一：经过类Thread
1 # 经过类Thread 2 import os 3 import time 4 from threading import Thread,get_ident 5 def func(name): 6 time.sleep(0.1) 7 print('线程：%s，该线程的进程id为：%s，线程id为：%s' %(name,os.getpid(),get_ident())) # get_ident()为类get_ident中的方法，做用是获取线程id 8 for i in range(10): 9 t = Thread(target=func, args=(i,)) 10 t.start() 11 12 》》》》结果： 13 线程：3，该线程的进程id为：14140，线程id为：13352 14 线程：0，该线程的进程id为：14140，线程id为：1672 15 线程：1，该线程的进程id为：14140，线程id为：15808 16 线程：2，该线程的进程id为：14140，线程id为：10136 17 线程：7，该线程的进程id为：14140，线程id为：4508 18 线程：6，该线程的进程id为：14140，线程id为：15068 19 线程：4，该线程的进程id为：14140，线程id为：13532 20 线程：5，该线程的进程id为：14140，线程id为：10836 21 线程：9，该线程的进程id为：14140，线程id为：9520 22 线程：8，该线程的进程id为：14140，线程id为：15568
-
代码演练-方式二：面向对象自定义类
1 # 自定义类建立多线程 【继承Thread类，同时重些run方法，如需添加本身的属性或者说加入参数继承父类init方法】 2 import os 3 from threading import Thread, get_ident 4 5 6 class Mythread(Thread): 7 def __init__(self, args): 8 super().__init__() 9 self.args = args 10 11 def run(self): 12 """ 13 执行函数 14 :return: 15 """ 16 print(self.args) # 就能够调用本身自定义的属性了 17 print('in thread 子线程Id:', get_ident(), '进程Id:', os.getpid()) 18 19 20 print("父进程python解释器:", os.getppid()) 21 print('进程即执行py文件:', os.getpid()) 22 print('主线程:', get_ident()) 23 t_obj = Mythread('china') 24 t_obj.start() 25 ''' 26 结果： 27 父进程python解释器: 1168 28 进程及执行py文件: 2724 29 主线程: 7468 30 china 31 in thread 子线程Id: 8180 进程Id: 2724 32 '''
- 多线程与多进程效率对比：线程开闭切远远高于进程
-
代码演练
1 '''效率：多线程开闭切开销远远小于进程隔了大几百倍''' 2 3 def func(a): 4 a += 1 5 6 7 if __name__ == '__main__': 8 start = time.time() 9 t_lis = [] 10 for i in range(50): 11 t = Thread(target=func, args=(i,)) 12 t.start() 13 t_lis.append(t) 14 for t in t_lis:t.join() 15 print('主线程') 16 print('时间：%s' % str(time.time() - start)) 17 18 start = time.time() 19 t_lis = [] 20 for i in range(50): 21 t = Process(target=func, args=(i,)) 22 t.start() 23 t_lis.append(t) 24 for t in t_lis: t.join() 25 print('主进程') 26 print('时间：%s' % str(time.time() - start)) 27 28 ''' 29 效率测试： 30 主线程 31 时间：0.008229732513427734 32 主进程 33 时间：5.307320833206177 34 '''
- 多线程数据共享：可以改变进程全局变量，不加锁，数据不安全
-
代码演练
1 '''线程间数据共享''' 2 from threading import Thread 3 4 n = 100 # 全局变量，存在主进程内存空间中 5 6 7 def func(): 8 global n 9 n -= 1 10 print(n) 11 12 13 if __name__ == '__main__': 14 t_li = [] 15 for i in range(100): 16 t = Thread(target=func) # 始终在同一个内存空间中 17 t.start() 18 t_li.append(t) 19 for t in t_li: 20 t.join() 21 print("线程：", n) 22 ''' 23 结果： 24 n = 0 25 每一个线程中的n从99，开始递减 26 ''' 27 28 p_li = [] 29 for i in range(100): 30 p = Process(target=func,) # 分别建立100个独立的内存空间，且相互独立 31 p.start() 32 p_li.append(p) 33 for p in p_li: 34 p.join() 35 print('进程：',n) 36 ''' 37 结果： 38 n = 100 39 每一个进程中的n为99 40 '''
-
- 守护线程：随主线程运行结束而结束，同进程不同，进程是主进程代码结束 【线程不能手动关闭没有terminate方法，只能经过守护线程来关闭，或者用模块】
-
代码演练
1 '''守护线程等待主线程结束而结束， 2 # 主线程结束，必须等全部非守护线程结束，才能结束 3 # 主线程结束，进程就结束了，进程必须保证全部非守护线程结束才行''' 4 import os 5 import time 6 from threading import Thread 7 8 def f1(): 9 print(True) 10 time.sleep(0.5) 11 print(os.getpid()) 12 13 def f2(): 14 print('in f2 start') 15 time.sleep(3) 16 print('in f2 end') 17 print(os.getpid()) 18 19 t = Thread(target=f1) 20 t.setDaemon(True) 21 t.start() 22 23 t2 = Thread(target=f2) 24 t2.start() 25 print('主线程',os.getpid()) 26 27 ''' 28 True 29 主线程 1440 30 in f2 start 31 1440 32 in f2 end 33 1440 34 '''
-
- Thread类方法汇总
- t.is_alive()或者t.isAlive() # 查看主线程是否存活
- t.getName() # 放回线程名如： Thread-1
- t.setName() # 设置线程名
- t.start() # 建立线程，调用run方法，运行
- t.join() # 主线程等待t线程执行结束后才继续，阻塞
- threading.currentTread() # 返回当前的线程变量，即对象： <_MainThread(MainThread, started 4644)>
- threading.enumerate() # 返回正在运行的线程对象列表，列表中线程对象确定包括主线程+开启的子线程
- threading.activeCount() # 返回正在运行的线程数，即 len（threading.enumerate() ）
- 锁和GIL: 互斥锁（也叫同步锁）、递归锁、全局解释器锁
- 在多个进程或线程同时访问一个数据的时候就会产生数据不安全的现象
- 只要多进程/多线程用到全局变量，就必须加锁，保证数据安全
- GIL锁：是同一个进程里的每个线程同一时间只能有一个线程访问cpu，锁的是线程，而不是具体的内存，但并不妨碍多个进程同时访问数据，这将致使访问同一数据不安全现象
- 递归锁：python支持同一进程/线程屡次请求同一资源，从而使资源能够被屡次require，直到一个线程全部的require都被release，其余线程才能得到资源。
- 互斥锁：同一时刻，只容许一个进程/线程访问资源，访问前加锁require阻塞，访问处理完毕后release，其余进程/线程才能访问，实现同步，保证数据安全。
- 为何要有GIL锁？
- 在开发python解释器的时候能够减小不少细粒度的锁
- 互斥锁与递归锁那个好？
- 一、递归锁，能快速解决死锁问题，快速恢复服务
- 二、但死锁问题的出现，是程序的设计或逻辑的问题，还应进一步的排除和重构逻辑来保证使用互斥锁也不会发生死锁问题
- 什么是死锁？
- 白话：多把锁同时应用在多个线程中
- 两个共享的数据资源，两个进程或线程各拿到了一个资源，但未同时拿到两个资源，都不释放，这时就会产生死锁现象，还有就是前面require了，后面忘记release了。
- 如何解决死锁？
- 若是在服务阶段 -> 递归锁快速恢复服务 ->排查逻辑 -> 互斥锁
- 若是在测试阶段 -> ->排查逻辑 -> 互斥锁
- 互斥锁和递归锁的区别？
- 互斥锁：就是在一个线程中不能连续屡次acquire
- 递归锁：能够在同一线程中acquire任意次，但同时注意相应的release次数要与acquire次数相同
- from multiprocessing/threading import RLock 导入递归锁
- from multiprocessing/threading import Lock 导入互斥锁
-
科学家吃面问题-互斥锁 【终极缘由是两把锁，可能会致使不一样的人各拿了一把锁，从而致使程序卡主，发生死锁现象】
场景： 多我的围着桌子吃一盘面，必须拿到叉子和面才能吃 若是一我的拿到了叉子，另外一我的拿到了面，就不能吃，就会致使僵直在一块儿 import time from threading import Thread,Lock lock = Lock() noodle_lock = Lock() fork_lock = Lock() def eat1(name): noodle_lock.acquire() print('%s拿到了面' % name) fork_lock.acquire() print('%s拿到了叉子' % name) print('%s在吃面'%name) time.sleep(0.5) fork_lock.release() # 0.01 noodle_lock.release() # 0.01 def eat2(name): fork_lock.acquire() # 0.01 print('%s拿到了叉子' % name) # 0.01 noodle_lock.acquire() print('%s拿到了面' % name) print('%s在吃面'%name) time.sleep(0.5) noodle_lock.release() fork_lock.release() eat_lst = ['alex','wusir','太白','yuan'] for name in eat_lst: # 8个子线程 7个线程 3个线程eat1,4个线程eat2 Thread(target=eat1,args=(name,)).start() Thread(target=eat2,args=(name,)).start() 结果： china拿到了面 china拿到了叉子 china在吃面 china拿到了叉子 usa拿到了面 --------程序开始卡死
-
递归锁解决-同一进程容许屡次require
1 # 递归锁解决死锁问题 2 import time 3 from threading import Thread, RLock 4 5 lock = RLock() 6 7 8 def eat1(name): 9 lock.acquire() 10 print('%s拿到了面' % name) 11 lock.acquire() 12 print('%s拿到了叉子' % name) 13 print('%s在吃面' % name) 14 time.sleep(0.5) 15 lock.release() # 0.01 16 lock.release() # 0.01 17 18 19 def eat2(name): 20 lock.acquire() # 0.01 21 print('%s拿到了叉子' % name) # 0.01 22 lock.acquire() 23 print('%s拿到了面' % name) 24 print('%s在吃面' % name) 25 time.sleep(0.5) 26 lock.release() 27 lock.release() 28 29 30 eat_lst = ['china', 'beijing', 'shanghai', 'shenzhen'] 31 for name in eat_lst: # 8个子线程 7个线程 3个线程eat1,4个线程eat2 32 Thread(target=eat1, args=(name,)).start() 33 Thread(target=eat2, args=(name,)).start()
-
互斥锁解决-更改设计逻辑，同时拿到叉子和面
1 import time 2 from threading import Thread,Lock 3 lock = Lock() 4 def eat1(name): 5 lock.acquire() 6 print('%s拿到了面' % name) 7 print('%s拿到了叉子' % name) 8 print('%s在吃面'%name) 9 time.sleep(0.5) 10 lock.release() # 0.01 11 12 def eat2(name): 13 lock.acquire() # 0.01 14 print('%s拿到了叉子' % name) # 0.01 15 print('%s拿到了面' % name) 16 print('%s在吃面'%name) 17 time.sleep(0.5) 18 lock.release() 19 20 eat_lst = ['china', 'beijing', 'shanghai', 'shenzhen'] 21 for name in eat_lst: # 8个子线程 7个线程 3个线程eat1,4个线程eat2 22 Thread(target=eat1,args=(name,)).start() 23 Thread(target=eat2,args=(name,)).start() 24 25 26 》》》： 27 china拿到了面 28 china拿到了叉子 29 china在吃面 30 china拿到了叉子 31 china拿到了面 32 china在吃面 33 beijing拿到了面 34 beijing拿到了叉子 35 beijing在吃面 36 beijing拿到了叉子 37 beijing拿到了面 38 beijing在吃面 39 shanghai拿到了面 40 shanghai拿到了叉子 41 shanghai在吃面 42 shanghai拿到了叉子 43 shanghai拿到了面 44 shanghai在吃面 45 shenzhen拿到了面 46 shenzhen拿到了叉子 47 shenzhen在吃面 48 shenzhen拿到了叉子 49 shenzhen拿到了面 50 shenzhen在吃面
- 信号量：同进程用法同样，池的效率高于信号量，不管进程，线程 【锁 + 计数器】
-
代码演练
1 # 进程中建立信号量,与开启进程池效率对比 2 # 池效率高于信号量 3 def ktv1(sem, i): 4 sem.acquire() 5 i += 1 6 sem.release() 7 8 9 def ktv2(i): 10 i += 1 11 12 13 # process 14 if __name__ == '__main__': 15 sem = Semaphore(5) # 同时只执行5个任务 16 start_time = time.time() 17 p_list = [] 18 for i in range(20): # 开启20个任务 19 p = Process(target=ktv1, args=(sem, i)) # 开启进程20个 20 p.start() 21 p_list.append(p) 22 for p in p_list: p.join() 23 print('process_semaphore:', time.time() - start_time) 24 25 pool = Pool(5) # 开启进程池，同一时间执行5个任务 26 start_time1 = time.time() 27 pool_list = [] 28 for i in range(20): # 开启20个任务 29 ret = pool.apply_async(func=ktv2, args=(i,)) # 异步 30 pool_list.append(ret) 31 pool.close() # 关闭进程池，再也不受理任务 32 pool.join() # 等待全部进程池中的任务结束 33 print('process_pool:', time.time() - start_time1) 34 ''' 35 process_semaphore: 2.2986388206481934 36 process_pool: 0.5303816795349121 37 ''' 38 39 ========================================== 40 # thread 类中没有线程池 41 # 但concurrent.futures中有 42 from threading import Thread,Semaphore,currentThread 43 from concurrent.futures import ThreadPoolExecutor 44 def f(sem,i): 45 sem.acquire() 46 i += 1 47 # print("sem",currentThread().getName()) # 获取线程名 48 sem.release() 49 50 def f2(i): 51 i += 1 52 # print("pool",currentThread().getName()) 53 54 start = time.time() 55 t_sem = Semaphore(5) # 线程信号量， 同时5个任务 56 t_list = [] 57 for i in range(20): # 20个任务，开启20个线程 58 t = Thread(target=f, args=(t_sem, i)) 59 t.start() 60 t_list.append(t) 61 for t in t_list:t.join() 62 print("in thread sem:" , time.time() - start) 63 64 start = time.time() 65 t_pool = ThreadPoolExecutor(5) 66 67 for i in range(20): 68 ret = t_pool.submit(f2,i) 69 t_pool.shutdown() 70 end = time.time() 71 print("in thread pool:", end- start) 72 ''' 73 in thread: 0.00498652458190918 74 in thread pool: 0.001001596450805664 75 '''
-
- 事件：event，同进程用法同样
-
代码演练-经过事件来实现检测网络的状态，从而决定是否链接数据库
1 '''事件：event，一个任务依赖另外一个任务的状态才进行下一步 2 wait 等待事件内部的信号变成True就不阻塞了 3 set 将标志改成True 4 clear 改为False 5 is_set 查看标志是否为True 6 ''' 7 # 数据库链接 8 import time 9 import random 10 from threading import Event,Thread 11 12 13 def check(e): 14 '''检测是否可以连通数据库，网络''' 15 print('正在检测两台机器之间的网络状况……') 16 time.sleep(random.randint(2,5)) 17 e.set() # 改为True，非阻塞 18 19 20 def connet_db(e): 21 print("status:", e.is_set()) 22 e.wait() 23 print("status:", e.is_set()) 24 print('链接数据库……') 25 print('链接数据库成功~~~') 26 27 # e = Event() 28 # Thread(target=connet_db, args=(e,)).start() 29 # Thread(target=check, args=(e,)).start() 30 ''' 31 status: False 32 正在检测两台机器之间的网络状况…… 33 status: True 34 链接数据库…… 35 链接数据库成功~~~ 36 ''' 37 38 39 def check(e): 40 '''检测是否可以连通数据库，网络''' 41 print('正在检测两台机器之间的网络状况……') 42 time.sleep(random.randint(2,5)) 43 e.set() # 改为True，非阻塞 44 45 def connet_db(e): 46 '''3次链接不上就退出''' 47 n = 0 48 while n < 3: 49 if e.is_set(): 50 break # 退出循环，执行链接库 51 else: 52 e.wait(timeout=0.5) 53 n += 1 54 if n == 3: 55 raise TimeoutError 56 print('链接数据库……') 57 print('链接数据库成功~~~') 58 59 60 e = Event() 61 Thread(target=connet_db, args=(e,)).start() 62 Thread(target=check, args=(e,)).start() 63 64 ''' 65 正在检测两台机器之间的网络状况…… 66 Exception in thread Thread-1: 67 Traceback (most recent call last): 68 File "D:\install\Python36\lib\threading.py", line 916, in _bootstrap_inner 69 self.run() 70 File "D:\install\Python36\lib\threading.py", line 864, in run 71 self._target(*self._args, **self._kwargs) 72 File "D:/install/project/七、并发编程/十一、threading_信号量.py", line 140, in connet_db 73 raise TimeoutError 74 TimeoutError 75 '''
-
- 线程池：threading 中没有数据池的类实现，数据池是另外一个包concurrent.futures中ThreadPoolExecutor实现，这个包一样能够启用进程池
- 队列：数据安全，先进先出，自带锁
-
几种锁
1 # from multiprocessing import Queue,JoinableQueue # 进程IPC队列 2 from queue import Queue # 线程队列 先进先出的 3 from queue import LifoQueue # 后进先出的 4 #方法： put get put_nowait get_nowait full empty qsize 5 # 队列Queue 6 # 先进先出 7 # 自带锁 数据安全 8 # 栈 LifoQueue 9 # 后进先出 10 # 自带锁 数据安全 11 # lq = LifoQueue(5)
- 条件：使得线程等待，只有知足某条件时，才释放n个线程
-
1 Python提供的Condition对象提供了对复杂线程同步问题的支持。 2 Condition被称为条件变量，除了提供与Lock相似的acquire和release方法外，还提供了wait和notify方法。 3 线程首先acquire一个条件变量，而后判断一些条件。 4 若是条件不知足则wait； 5 若是条件知足，进行一些处理改变条件后，经过notify方法通知其余线程，其余处于wait状态的线程接到通知后会从新判断条件。不断的重复这一过程，从而解决复杂的同步问题。
-
代码演练
1 from threading import Condition 2 # acquire 3 # release 4 # wait 阻塞 5 # notify 让wait解除阻塞的工具 6 # wait仍是notify在执行这两个方法的先后 必须执行acquire和release 7 from threading import Condition,Thread 8 def func(con,i): 9 con.acquire() 10 # 判断某条件 11 con.wait() 12 print('threading : ',i) 13 con.release() 14 15 con = Condition() 16 for i in range(20): 17 Thread(target=func,args=(con,i)).start() 18 con.acquire() 19 # 帮助wait的子线程处理某个数据直到知足条件 20 con.notify_all() 21 con.release() 22 while True: 23 num = int(input('num >>>')) 24 con.acquire() 25 con.notify(num) 26 con.release()
-
- 定时器：阻塞 定时器，指定n秒后执行某个操做
-
代码演练
1 from threading import Timer 2 3 4 def func(): 5 print('执行我啦') 6 7 8 # interval 时间间隔 9 Timer(0.2, func).start() # 定时器 10 # 建立线程的时候,就规定它多久以后去执行
-
-
内置包：concurrent.futures 高度封装，管理数据池【可建进程/线程池】！！！
-
1 #1 介绍 2 concurrent.futures模块提供了高度封装的异步调用接口 3 ThreadPoolExecutor：线程池，提供异步调用 4 ProcessPoolExecutor: 进程池，提供异步调用 5 Both implement the same interface, which is defined by the abstract Executor class. 6 7 #2 基本方法 8 #submit(fn, *args, **kwargs) 9 异步提交任务 10 11 #map(func, *iterables, timeout=None, chunksize=1) 12 取代for循环submit的操做 13 14 #shutdown(wait=True) 15 至关于进程池的pool.close()+pool.join()操做 16 wait=True，等待池内全部任务执行完毕回收完资源后才继续 17 wait=False，当即返回，并不会等待池内的任务执行完毕 18 但无论wait参数为什么值，整个程序都会等到全部任务执行完毕 19 submit和map必须在shutdown以前 20 21 #result(timeout=None) 22 取得结果 23 24 #add_done_callback(fn) 25 回调函数 - 总结：
- 一、并发：每一个线程共用进程ID（因此共享进程资源），每一个子线程拥有本身的pid，经过get_ident()【需导入类get_ident】，由于GIL的存在，多线程没法并行。
- 二、数据共享：多线程之间数据共享，可改变主进程全局变量。
- 三、效率：多线程开闭，切时间开销远远小于多进程
- 四、守护线程：主线程结束，守护线程结束，（主线程结束的条件是全部非守护线程结束）故，守护线程结束，意味着全部线程都结束了
- 五、守护进程：主进程代码结束，守护进程结束，此时非守护进程并不必定都结束了，但主进程会等全部非守护进程结束才会结束，从而回收资源。
- 六、为何还要用GIL：GIL为cpython解释器独有，历史遗留，但开发python解释器时可减小不少细粒度的锁
- 七、开启多线程会出现主线程ID（执行整个文件代码 get_ident()）,进程ID，子线程ID
- 八、线程开启不用放在 if__name__== ‘__main__’ 下，由于共享同一个进程资源，进程开启需求，由于每一个进程会另开新的内存空间，同时将主进程变量都导入执行。
- 九、解决并发三个工具就能解决大部分问题：join() 同步控制，用户获取结果，锁 保证数据安全，池 提升效率，解决并发
- 十、进程池中的回调函数在主进程运行，线程池中的回调函数在子线程运行。
-
进程/线程中各类对比*****
- 进程/线程中的信号量和线程/进程池的区别？
- 信号量/进程池都是同一时刻容许固定数量的进程/线程访问修改数据，实现并发效果。
- 信号量是有多少个任务，就开启多少个进程/线程，进程/线程池是不管任务多少，都只开启指定数量的进程。
- 信号量减小了操做系统进程/线程切换的负担，但增长了进程/线程开启，关闭的时间开销，进程/线程池是进程/线程开启后并不因任务结束而关闭，而是等待新任务，这样就减小了进/线程开闭的时间开销，也减少了操做系统的调度压力。
- 从效率上看，进程池的效率要远远高于信号量开启多进程的效率。
- 注：信号量100我的开100个任务，同时只能干5个任务，进程池100个任务，5我的干，同时也只能干5个任务
- 何时用多进程，何时用多线程？
- 开多进程/线程都是为了实现高并发
- 当任务是高并发计算型的任务，为了充分利用cpu，使用多进程较好
- 当任务为网络/文件/数据库等IO操做时，使用多线程较好
- 当任务为既有高并发计算型又有少许IO操做的，能够启用多进程和多线程一块儿较好
- 进程与线程的区别？
- 线程不能独立存在，必须在一个进程里
- 线程的开闭及切换的时间开销远远小于进程
- 同一个进程间的多个线程数据共享，多进程间异步，数据隔离不共享
- cpython解释器中全局解释器锁CIL的存在，致使同一进程多个线程同一时刻只能一个线程访问cpu，从而不能充分利用多核处理器，而多进程就没有这个限制
- 进程/线程中的信号量和线程/进程池的区别？
4、python中的并发编程-协程
-
- 什么是协程？
- 是一种用户态的轻量级线程，实现单线程下的并发，由用户程序而非操做系统调度控制，在一个线程内的多个任务之间互相切换。
- python中最简单的切换以及状态的保存，使用原生的yield就是协程最基本的体现。【yeild 只有程序之间的切换,没有重利用任何IO操做的时间】
- 通常状况下都是经过gevent这个模块来完成协程的工做，它内部使用的协程机制是greenlnet，它可以帮助咱们规避IO操做，最大限度的利用cpu。
- 协程比起线程的区别和优点？
- 区别：
- 一、线程的切换是由操做系统控制调度的，协程的切换是由用户级别自行控制调度的。
- 二、线程开闭，切换有时间开销，协程切换，开闭的开销更小，可经过gevent实现自动切换从而无时间开销（由于协程是在同一线程下进行的，任务之间的切换是用户级的）
- 三、线程是在进程下开启的，协程是在线程下开启的
- 优点：
- 一、单线程内就就能够实现并发的效果，最大限度的理由CPU【优势】，协程程序级别的切换开销更小，操做系统彻底感知不到，于是更加轻量级【优势】。
- 二、协程能在一条线程的基础上，遇到IO在多个任务之间互相切换，节省了线程开启的开销，充分的利用了一条线程来提升cpu的工做效率。
- 三、协程不存在数据不安全的问题（由于协程是遇IO切，同一刻，只有一个任务运行，同步效果）
- 四、程序不会由于协程中某一个任务进入阻塞状态而使整条线程阻塞。
- 缺点：
- 一、本质是单线程下，没法利用多核
- 二、协程一旦出现阻塞，将会阻塞整条线程
- 特色：
- 一、必须在只有一个单线程里实现并发
- 二、修改共享数据不须要枷锁
- 三、用户程序里本身保存多个控制流的上下文栈
- 四、遇到IO就切换，原生协程自能手动切
- 五、gevent模块select机制能够帮助实现监测IO，使一个协程遇到IO操做自动切换到其余协程
- 六、yield，greenlet都没法实现监测IO，自动切换，遇到阻塞仍是在阻塞。
相关文章
- 1. Python学习day35-并发编程(1)
- 2. Python学习day40-并发编程(终)
- 3. Python学习day37-并发编程(3)
- 4. java并发编程学习
- 5. 学习Java并发编程
- 6. java并发编程-学习
- 7. Java 并发编程（学习）
- 8. python-学习-python并发编程之多进程与多线程
- 9. Python并发编程
- 10. python并发编程
- 更多相关文章...
- • Rust 并发编程 - RUST 教程
- • 您已经学习了 XML Schema，下一步学习什么呢？ - XML Schema 教程
- • Tomcat学习笔记（史上最全tomcat学习笔记）
- • 适用于PHP初学者的学习线路和建议
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. Python学习day35-并发编程(1)
- 2. Python学习day40-并发编程(终)
- 3. Python学习day37-并发编程(3)
- 4. java并发编程学习
- 5. 学习Java并发编程
- 6. java并发编程-学习
- 7. Java 并发编程（学习）
- 8. python-学习-python并发编程之多进程与多线程
- 9. Python并发编程
- 10. python并发编程