一、什么是进程
进程（Process）是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。在当代面向线程设计的计算机结构中，进程是线程的容器。程序是指令、数据及其组织形式的描述，进程是程序的实体。
狭义定义：进程是正在运行的程序的实例（an instance of a computer program that is being executed）。 广义定义：进程是一个具有一定独立功能的程序关于某个数据集合的一次运行活动。 进程理论 程序:一堆死代码 进程:正在运行的程序 进程是资源单位,进程与进程之间数据是绝对意义上的物理隔离,但是可以利用某些技术实现数据交互(消息队列)
动态性：进程的实质是程序在多道程序系统中的一次执行过程，进程是动态产生，动态消亡的。 并发性：任何进程都可以同其他进程一起并发执行 独立性：进程是一个能独立运行的基本单位，同时也是系统分配资源和调度的独立单位； 异步性：由于进程间的相互制约，使进程具有执行的间断性，即进程按各自独立的、不可预知的速度向前推进 结构特征：进程由程序、数据和进程控制块三部分组成。 多个不同的进程可以包含相同的程序：一个程序在不同的数据集里就构成不同的进程，能得到不同的结果；但是执行过程中，程序不能发生改变
注意：同一个程序执行两次，就会在操作系统中出现两个进程，所以我们可以同时运行一个软件，分别做不同的事情也不会混乱
二、进程的调度
要想多个进程交替运行，操作系统必须对这些进程进行调度，这个调度也不是随即进行的，而是需要遵循一定的法则，由此就有了进程的调度算法
# 先来先服务调度算法
先来先服务（FCFS）调度算法是一种最简单的调度算法，该算法既可用于作业调度，也可用于进程调度。FCFS算法比较有利于长作业（进程），而不利于短作业（进程）。由此可知，本算法适合于CPU繁忙型作业，而不利于I/O繁忙型的作业（进程）
# 短作业优先调度算法
短作业（进程）优先调度算法（SJ/PF）是指对短作业或短进程优先调度的算法，该算法既可用于作业调度，也可用于进程调度。但其对长作业不利；不能保证紧迫性作业（进程）被及时处理；作业的长短只是被估算出来的。
# 时间片轮转法
时间片轮转(Round Robin，RR)法的基本思路是让每个进程在就绪队列中的等待时间与享受服务的时间成比例。在时间片轮转法中，需要将CPU的处理时间分成固定大小的时间片，例如，几十毫秒至几百毫秒。如果一个进程在被调度选中之后用完了系统规定的时间片，但又未完成要求的任务，则它自行释放自己所占有的CPU而排到就绪队列的末尾，等待下一次调度。同时，进程调度程序又去调度当前就绪队列中的第一个进程
在轮转法中，加入到就绪队列的进程有3种情况：
一种是分给它的时间片用完，但进程还未完成，回到就绪队列的末尾等待下次调度去继续执行。
另一种情况是分给该进程的时间片并未用完，只是因为请求I/O或由于进程的互斥与同步关系而被阻塞。当阻塞解除之后再回到就绪队列。
第三种情况就是新创建进程进入就绪队列。
如果对这些进程区别对待，给予不同的优先级和时间片从直观上看，可以进一步改善系统服务质量和效率。例如，我们可把就绪队列按照进程到达就绪队列的类型和进程被阻塞时的阻塞原因分成不同的就绪队列，每个队列按FCFS原则排列，各队列之间的进程享有不同的优先级，但同一队列内优先级相同。这样，当一个进程在执行完它的时间片之后，或从睡眠中被唤醒以及被创建之后，将进入不同的就绪队列。
# 多级反馈队列调度算法
多级反馈队列调度算法不必事先知道各种进程所需的执行时间，而且还可以满足各种类型进程的需要，因而它是目前被公认的一种较好的进程调度算法。
三、并行与并发
# 并行 : 并行是指两者同时执行，比如赛跑，两个人都在不停的往前跑；（资源够用，比如三个线程，四核的CPU ） # 并发 : 并发是指资源有限的情况下，两者交替轮流使用资源，比如一段路(单核CPU资源)同时只能过一个人，A走一段后，让给B，B用完继续给A ，交替使用，目的是提高效率。 # 区别: 并行是从微观上，也就是在一个精确的时间片刻，有不同的程序在执行，这就要求必须有多个处理器。 并发是从宏观上，在一个时间段上可以看出是同时执行的，比如一个服务器同时处理多个session。
四、同步、异步、阻塞、非阻塞
同步与异步:描述的是任务的提交方式
同步:提交任务之后原地等待任务的返回结果
异步:提交任务之后 继续执行的后续代码 不等待结果 （回调机制）
阻塞与非阻塞:描述的是程序的运行状态
阻塞:程序阻塞态
非阻塞:就绪态或者运行态
在程序运行的过程中，由于被操作系统的调度算法控制，程序会进入几个状态：就绪，运行和阻塞。
# 同步与异步 所谓同步就是一个任务的完成需要依赖另外一个任务时，只有等待被依赖的任务完成后，依赖的任务才能算完成，这是一种可靠的任务序列 所谓异步是不需要等待被依赖的任务完成，只是通知被依赖的任务要完成什么工作，依赖的任务也立即执行，只要自己完成了整个任务就算完成了。至于被依赖的任务最终是否真正完成，依赖它的任务无法确定，所以它是不可靠的任务序列 # 阻塞与非阻塞 阻塞和非阻塞这两个概念与程序（线程）等待消息通知(无所谓同步或者异步)时的状态有关。也就是说阻塞与非阻塞主要是程序（线程）等待消息通知时的状态角度来说的
# 同步阻塞形式：效率最低 # 异步阻塞形式：采用的是异步的方式去等待消息被触发 异步操作是可以被阻塞住的，只不过它不是在处理消息时阻塞，而是在等待消息通知时被阻塞 # 同步非阻塞形式：效率低 程序需要在这两种不同的行为之间来回的切换 # 异步非阻塞形式：效率更高 程序没有在两种不同的操作中来回切换
五、进程：multiprocessing模块
Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动） 强调： 1. 需要使用关键字的方式来指定参数 2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号 参数介绍： 1 group参数未使用，值始终为None 2 target表示调用对象，即子进程要执行的任务 3 args表示调用对象的位置参数元组，args=(1,2,'egon',) 4 kwargs表示调用对象的字典,kwargs={'name':'egon','age':18} 5 name为子进程的名称 1 p.start()：启动进程，并调用该子进程中的p.run() 2 p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法 3 p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁 4 p.is_alive():如果p仍然运行，返回True 5 p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程 1 p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置 2 p.name:进程的名称 3 p.pid：进程的pid 4 p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可) 5 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）
1.进程的创建与结束
进程是资源单位，进程之间数据隔离
# 第一种方式
from multiprocessing import Process import time def task(name): print('%s is running'%name) time.sleep(3) print('%s is end'%name) # windows操作中创建的方式 会将以模块导入的方式从头到尾执行当前文件 # linux中 是直接当前文件的代码拷贝一份(fork) if __name__ == '__main__': # 双下name p = Process(target=task,args=('jason',)) # 函数名加括号优先级最高 p.start() # 告诉操作系统帮你创建一个进程:1.申请内存空间 2.执行当前文件中的代码 将产生的名字全部丢到新申请的名称空间中 print('主')
# 创建进程的第二种方式 from multiprocessing import Process import time class MyProcess(Process): def __init__(self,name): super().__init__() self.name = name def run(self): # 第二种方式里面的run就类似于第一种方式里面的task print('%s is running'%self.name) time.sleep(1) print('%s is end'%self.name) if __name__ == '__main__': p = MyProcess('jason') p.start() print('主')
2、进程join的方法(按要求顺序执行)
from multiprocessing import Process import time def task(name): print('%s is running' % name) time.sleep(3) print('%s is end'%name) if __name__ == '__main__': p = Process(target=task,args=('simon',))
# p = Process(target=task,kwargs={'name':'jason'}) # 传参的第二种方式 p.start() p.join() # 主进程等待指定的子进程运行结束之后才执行 不影响其他子进程的运行 print("主") """ simon is running simon is end 主 """
from multiprocessing import Process import time def task(name,n): print('%s is running' % name) time.sleep(n) print('%s is end'%name) if __name__ == '__main__': start_time = time.time() p_list = [] for i in range(1,4): p = Process(target=task,args=('子进程%s'%i,i**i)) p.start() p_list.append(p) for p in p_list: p.join() print('主') print('耗时:%s'%(time.time() - start_time)) """ 子进程1 is running 子进程2 is running 子进程3 is running 子进程1 is end 子进程2 is end 子进程3 is end 主 耗时:27.12771248817444 """
3.验证进程之间的数据隔离
from multiprocessing import Process money = 10000000000000 def task(): global money money = 0 print('我真的运行了 没骗你') if __name__ == '__main__': p = Process(target=task) p.start() p.join() print(money) """ 我真的运行了 没骗你 10000000000000 """
4.进程对象及其它方法
1.current_process查看进程号 2.os.getpid() 查看进程号 os.getppid() 查看父进程进程号 3.进程的名字，p.name直接默认就有，也可以在实例化进程对象的时候通过关键字形式传入name='' 4.p.terminate() 杀死子进程 5.p.is_alive() 判断进程是否存活 3，4结合看不出结果，因为操作系统需要反应时间。主进程睡0.1即可看出效果
from multiprocessing import Process,current_process import time import os def task(): # print('%s is running'%current_process().pid) # 查看当前进程的id号 # print('%s is running'%os.getpid()) # 查看当前进程的id号 print('%s is running'%os.getppid()) # 查看当前进程的父进程的id号 # print('%s is running'%Process().pid) # 查看当前进程的id号 time.sleep(3) # print('%s is end'%current_process().pid) print('%s is end'%os.getpid()) if __name__ == '__main__': p = Process(target=task) p.start() print(p.name) p.terminate() # 杀死子进程 time.sleep(0.1) # 操作系统杀死子进程也需要一定时间 # 判断当前子进程是否存活 print(p.is_alive()) # False # print('主:%s'%current_process().pid) # print('主:%s'%os.getpid()) print('主:%s'%os.getppid()) print(current_process().name)
"""
Process-1
False
主:12708
MainProcess
"""
5.守护进程
# 主进程死子进程立马就死 # p.daemon = True # 必须加在p.start()之前 加之后会报错 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设置为当前主进程的守护进程 p.start() # p.daemon = True # 一定要写在p.start之前,否则就会报错AssertionError: process has already started time.sleep(1) print('皇帝jason驾崩了')
6.僵尸进程和孤儿进程
引子:进程与进程之间是相互独立的，但是为什么主进程还要等子进程结束才会结束呢？ 所有的子进程在运行结束后都会变成僵尸进程，还保留着pid及占用cpu多长时间等信息。这些信息会被主进程主动回收(两种回收可能:1.主进程调用join的等待子进程结束回收。2.主进程正常结束调用wait回收) 孤儿进程：子进程还没结束，主进程先死了。意味着子进程的pid等信息不能被及时回收。要等到init回收机制来帮忙回收，会稍微熬时长一点
孤儿进程无害，因为有init管
僵尸进程：当父进程永远不死 并且不调用join方法回收子进程的资源 还在源源不断的创建新的子进程 僵尸进程有害，父进程不死并且一直不停的在创建新的子进程
7.互斥锁
# 多个进程同一时间操作同一份数据的时候会造成数据的错乱 如何解决这一现象? 锁:互斥锁 互斥锁特点 1.牺牲了程序的效率但是保证了数据的安全 # ps:你自己千万不要随意处理锁的问题, 处理不得当会造成死锁现象 互斥锁应该在主进程中产生 交给子进程去使用 # 互斥锁将并发变成串行，牺牲效率保证数据的安全性，通常用于对数据的修改部分代码上
# 以抢票来进行模拟：多进程同时操作数据 # json格式文件内容：{"ticket": 1} """ # 思路： 1.查票 2.买票 1.查票 2.买票 """ from multiprocessing import Process,Lock import json import time import random def check(name): with open('ticket',encoding='utf-8') as f: data = json.load(f) print('当前余票:%s'%data.get("ticket")) def buy(name): with open('ticket',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('ticket','w',encoding='utf-8') as f: json.dump(data,f) print('%s抢票成功'%name) else: print('没票了 回不了家了!') def run(name,mutex): check(name) mutex.acquire() # 抢锁 buy(name) mutex.release() # 释放锁 if __name__ == '__main__': mutex = Lock() # 产生了一把 for i in range(10): p = Process(target=run,args=('子进程:%s'%i,mutex)) p.start() # json格式文件：{"ticket": 0}
8.进程间通信(IPC机制)
# 先来了解队列的概念 先进先出 from multiprocessing import Queue q = Queue(5) # 实例化一个q对象，参数可传可不传，用来限制队列的大小 # 步骤1；for循环放，演示超出范围阻塞情况 for i in range(6): q.put(i) # 朝队列里放值 # 步骤2:手动一个个放 q.put(1~5) q.full() # 查看当前队列是否已满 q.get() # 获取队列中的值，取多了也会阻塞住 q.empty() # 判断队列是否取空了 # 两个地方需要注意，队列满了的时候，再放值会阻塞，直到前面的值被取走 # 队列空了的时候，再取值也会阻塞，直到有人再往这里面放值 q.get_nowait() # 有值就拿，没值报错 """ 进程间使用队列传输数据 """ # 步骤1:子进程生成数据，主进程获取打印 # 步骤2:子进程生成数据，另一个子进程获取数据 from multiprocessing import Queue,Process def producer(q): q.put('hello big baby!') def consumer(q): print(q.get()) if __name__ == '__main__': q = Queue() p = Process(target=producer,args=(q,)) p.start() p1 = Process(target=consumer,args=(q,)) p1.start()
9.生产者消费者模型
日常生活中买包子示例:(包子做的多了买的人少，包子做的少了买的人多)
程序中示例(一个程序负责生产数据，一个程序负责消费数据)
如何解决:弄一个存储数据的容器，满了就不再生产，不满就继续生产，无论生产的多还是消费的多，都可以很好的解决，生产的多就多开几个消费者进程，消费的多就多开几个生产者进程
总的来说就是如何解决生产与消费之间平衡的事情
# 生产数据和消费数据都模拟一下延迟 # 先写两个生产者，生产数据, 再写一个消费者消费数据 # 再供需平衡 二对二 # 抛出问题 程序都最好生产者生产完消费者也消费完。还不结束如何处理? # 主进程等待生产者的进程结束，然后往队列里丢一个None运行还是卡住 # 诠释队列中的数据多进程来获取是数据安全的，即对进程获取数据是加锁处理 # 主进程放两个None达到效果 # JoinableQueue使用 from multiprocessing import Process,Queue,JoinableQueue import time import random def producer(name,food,q): for i in range(3): data = '%s生产了%s%s'%(name,food,i) time.sleep(random.randint(1,3)) q.put(data) print(data) def consumer(name,q): while True: data = q.get() if data == None:break time.sleep(random.randint(1,3)) print('%s吃了%s'%(name,data)) q.task_done() # 已经从队里中拿到数据 并且处理完毕了 if __name__ == '__main__': q = JoinableQueue() p = Process(target=producer,args=('大厨jason','肉包子',q)) p1 = Process(target=producer,args=('配菜owen','馒头',q)) c = Process(target=consumer,args=('吃货egon',q)) c1 = Process(target=consumer,args=('坑货尚老师',q)) p.start() p1.start() c.daemon = True c1.daemon = True c.start() c1.start() p.join() p1.join() q.join() # 等带队列中所有的数据都被取干净
六、线程:threading模块
1.线程理论
1.什么是线程？ 进程:资源单位(起一个进程仅仅只是在内存中开辟一块空间) 线程:执行单位(真正被cpu执行的其实是线程，线程其实指的就是代码的执行过程,执行代码中需要的数据找进程这个资源单位要！) # 也就意味着进程中真正在执行功能的其实是它里面的线程，即每个进程内部都必须起码有一个线程 2.为什么要有线程？ # 1.同一个进程下的多个线程共享该进程内的资源 # 2.创建线程的开销要远远小于进程 开进程 1.申请内存空间 2.执行代码 开线程 上面的两个都没有 再来举例阐述 开发一个文本编辑器工具,功能获取用户输入并显示到屏幕上并适时存入硬盘 3.进程中自带一个线程 同一个进程下 多个线程数据是共享的(*****) 进程自带的那个线程我们会人为的叫它 主线程 但是同一个进程下 多个线程没有主次之分
2.创建线程的2种方式
# 第一种方式 from threading import Thread import time def task(name): print('%s is running'%name) time.sleep(3) print('%s is end'%name) t = Thread(target=task,args=('jason',)) t.start() # 创建线程 这句话一运行完 线程几乎就已经创建完毕了 print('主') # 第二种方式 from threading import Thread import time class MyThread(Thread): def __init__(self,name): super().__init__() self.name = name def run(self): print('%s is running'%self.name) time.sleep(3) print('%s is end'%self.name) t = MyThread('jason') t.start() print('主') """ jason is running 主 jason is end """
3.线程对象的属性和方法
主线程一定要等待所有子线程的结束才会结束
因为主线程一旦结束意味着整个进程的结束,那么会造成子线程无法正常运行
# 1.验证线程之间数据共享
# 同一个进程下多个线程数据是共享 from threading import Thread import time n=100 def task(): global n n=0 if __name__ == '__main__': t=Thread(target=task) t.start() t.join() print('主',n) # 2.线程之间os.getpid() >>> 相等 from threading import Thread import time,os def task(): print('%s is running' %os.getpid()) if __name__ == '__main__': t=Thread(target=task) t.start() print('主',os.getpid()) # 3.active_count当前活跃的线程个数,当前线程 from threading import Thread,active_count,current_thread import time,os def task(): print('%s is running' %current_thread().name) time.sleep(2) if __name__ == '__main__': t=Thread(target=task,) t.start() # t.join() # 主线程等待子线程结束 # print('主',active_count()) print('主',current_thread().name)
4.守护线程
# 同一个进程下 主线程会等待所有非守护线程的结束才会结束 from threading import Thread,current_thread import time def task(): print('%s is running'%current_thread().name) time.sleep(3) print('%s is end'%current_thread().name) t = Thread(target=task) #t1 = Thread(target=task) t.daemon = True t.start() #t1.start() print('主') """ Thread-1 is running 主 """
5.线程互斥锁
# 锁： 1.将并发变成串行 2.降低了程序的运行效率但是提高数据的安全性 # from multiprocessing import Process,Lock from threading import Thread,Lock import time n = 100 def task(mutex): global n mutex.acquire() # 抢锁 temp = n time.sleep(0.1) n = temp - 1 mutex.release() # 释放锁 if __name__ == '__main__': mutex = Lock() t_list = [] for i in range(100): t = Thread(target=task,args=(mutex,)) t.start() t_list.append(t) for t in t_list: t.join() print(n)
七、GIL全局解释器
结论：在Cpython解释器中，同一个进程下开启的多线程，同一时刻只能有一个线程执行，无法利用多核优势 1.GIL全局解释锁只在Cpython解释器上才有 2.GIL全局解释锁本质也是一把互斥锁 3.GIL全局解释锁用来锁主同一个进程下多个线程的运行 4.因为内存管理不是线程安全的 内存管理(垃圾回收机制) 1.引用计数 2.标记清除 3.分代回收 1.cpython解释器下python的同一个进程下的多线程真的一点用没有吗？ 你不能够直接下结论，应该分任务的类型来讨论 四个任务都是计算密集型的 10s 多线程 40s+ 多进程 10s+ 四个任务都是IO密集型的 10s 多进程 1.申请内存空间 2.执行代码 多线程(优势的) 当你的任务是以计算为主那么你可以开多进程 当你的任务是以IO为主的那么你可以开多线程
# 计算密集型 多进程 from multiprocessing import Process from threading import Thread import os,time def work(): res=0 for i in range(100000000): res*=i if __name__ == '__main__': l=[] print(os.cpu_count()) # 本机为6核 start=time.time() for i in range(12): p=Process(target=work) #耗时8.779496908187866s多 # p=Thread(target=work) #耗时44.838125467300415s多 l.append(p) p.start() for p in l: p.join() stop=time.time() print('run time is %s' %(stop-start)) # IO密集型 多线程 from multiprocessing import Process from threading import Thread import threading import os,time def work(): time.sleep(2) if __name__ == '__main__': l=[] print(os.cpu_count()) #本机为6核 start=time.time() for i in range(400): # p=Process(target=work) #耗时20.29075288772583s多,大部分时间耗费在创建进程上 p=Thread(target=work) #耗时 2.0506210327148438s多 l.append(p) p.start() for p in l: p.join() stop=time.time() print('run time is %s' %(stop-start))
八、进程池与线程池
无论是多进程还是多线程 你都不应该无限制开下去
因为这样会造成硬件层面上的损坏 而一旦硬件损坏了 你的代码再牛逼也没有用武之地
原因
为了保证计算机硬件能够正常运行
池:
降低了程序的运行效率但是保证了计算机硬件安全
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor import time import os # 示例化池对象 # 不知道参数的情况，默认是当前计算机cpu个数乘以5，也可以指定线程个数 # pool = ThreadPoolExecutor(20) # 创建了一个池子，池子里面有20个线程 pool = ProcessPoolExecutor(5) # 创建了一个池子 def task(n): print(n,os.getpid()) time.sleep(2) return n**2 def call_back(n): print('我拿到了结果:%s'%n.result()) """ 提交任务的方式 同步:提交任务之后，原地等待任务的返回结果，再继续执行下一步代码 异步:提交任务之后，不等待任务的返回结果(通过回调函数拿到返回结果并处理)，直接执行下一步操作 """ # 回调函数:异步提交之后一旦任务有返回结果，自动交给另外一个去执行 if __name__ == '__main__': # pool.submit(task,1) t_list = [] for i in range(20): future = pool.submit(task,i).add_done_callback(call_back) # 异步提交任务 t_list.append(future) # pool.shutdown() # 关闭池子并且等待池子中所有的任务运行完毕 # for p in t_list: # print('>>>:',p.result()) print('主')
九、互斥锁和递归锁
from threading import Thread,Lock,RLock import time # mutexA=Lock() # mutexB=Lock() mutexB=mutexA=RLock() # 递归锁 """ 递归锁自身有计数功能 他能够记住 被acquire了多少次 只有第一个抢到递归锁才能够连续的acquire连续的release """ class Mythead(Thread): def run(self): self.f1() self.f2() def f1(self): mutexA.acquire() print('%s 抢到A锁' %self.name) mutexB.acquire() print('%s 抢到B锁' %self.name) mutexB.release() mutexA.release() def f2(self): mutexB.acquire() print('%s 抢到了B锁' %self.name) time.sleep(2) mutexA.acquire() print('%s 抢到了A锁' %self.name) mutexA.release() mutexB.release() if __name__ == '__main__': for i in range(100): t=Mythead() t.start()