python 并发编程
2018-12-03 10:17:06 7 举报
AI智能生成
python并发编程 线程 进程 操作系统
作者其他创作
大纲/内容
操作系统
为什么要有操作系统
程序员无法把所有的硬件操作细节都了解到,管理这些硬件并且加以优化使用是非常繁琐的工作,这个繁琐的工作就是操作系统来干的, 有了他,程序员就从这些繁琐的工作中解脱了出来,只需要考虑自己的应用软件的编写就可以了,应用软件直接使用操作系统提供的功能来间接使用硬件。
什么是操作系统
精简的说的话,操作系统就是一个协调、管理和控制计算机硬件资源和软件资源的控制程序
操作系统所处的位置如图1
细说的话,操作系统应该分成两部分功能:
一: 隐藏了丑陋的硬件调用接口(键盘、鼠标、音箱等等怎么实现的,就不需要你管了),为应用程序员提供调用硬件资源的更好,更简单,更清晰的模型(系统调用接口)。应用程序员有了这些接口后,就不用再考虑操作硬件的细节,专心开发自己的应用程序即可。
二:将应用程序对硬件资源的竞态请求变得有序化
操作系统与普通软件的区别
1.主要区别是:你不想用暴风影音了你可以选择用迅雷播放器或者干脆自己写一个,但是你无法写一个属于操作系统一部分的程序(时钟中断处理程序),操作系统由硬件保护,不能被用户修改。
2.操作系统与用户程序的差异并不在于二者所处的地位。特别地,操作系统是一个大型、复杂、长寿的软件,
操作系统发展史
第一代(1940~1955) 手工操作----穿孔卡片
第二代(1955~1965) 磁带存储---批处理系统
第三代(1955~1965) 多道程序系统(多道是重点*****)
操作系统原理-内容包含
操作系统的资源管理技术
系统调用
操作系统内核
处理器状态: 内核态和用户态
中断(Interupt)
进程
处理器调度
进程的交互
临界区管理
信号量(samaphore)和PV操作
管程
死锁
可变分区存储管理
分页存储管理
分段存储管理
虚拟存储管理
请求段页式虚拟内存管理
I/O硬件原理:I/O控制方式
I/O软件原理
缓冲技术
设备独立性
虚拟设备
文件逻辑结构
文件物理结构
进程Process
背景知识
顾名思义,进程即正在执行的一个过程。进程是对正在运行程序的一个抽象。
进程的概念起源于操作系统,是操作系统最核心的概念,也是操作系统提供的最古老也是最重要的抽象概念之一。操作系统的其他所有内容都是围绕进程的概念展开的。
进程的概念起源于操作系统,是操作系统最核心的概念,也是操作系统提供的最古老也是最重要的抽象概念之一。操作系统的其他所有内容都是围绕进程的概念展开的。
必备的理论基础:
事先了解操作系统
1:隐藏丑陋复杂的硬件接口,提供良好的抽象接口
2:管理、调度进程,并且将多个进程对硬件的竞争变得有序
2:管理、调度进程,并且将多个进程对硬件的竞争变得有序
多道技术
1.产生背景:针对单核,实现并发
ps:
现在的主机一般是多核,那么每个核都会利用多道技术
有4个cpu,运行于cpu1的某个程序遇到io阻塞,会等到io结束再重新调度,会被调度到4个
cpu中的任意一个,具体由操作系统调度算法决定。
2.空间上的复用:如内存中同时有多道程序
3.时间上的复用:复用一个cpu的时间片
强调:遇到io切,占用cpu时间过长也切,核心在于切之前将进程的状态保存下来,这样
才能保证下次切换回来时,能基于上次切走的位置继续运行
复制代码
ps:
现在的主机一般是多核,那么每个核都会利用多道技术
有4个cpu,运行于cpu1的某个程序遇到io阻塞,会等到io结束再重新调度,会被调度到4个
cpu中的任意一个,具体由操作系统调度算法决定。
2.空间上的复用:如内存中同时有多道程序
3.时间上的复用:复用一个cpu的时间片
强调:遇到io切,占用cpu时间过长也切,核心在于切之前将进程的状态保存下来,这样
才能保证下次切换回来时,能基于上次切走的位置继续运行
复制代码
什么是进程
进程(Process)是计算机中的程序关于某数据集合上的一次运行活动,是系统进行资源分配和调度的基本单位,是操作系统结构的基础。
狭义定义
进程是正在运行的程序的实例(an instance of a computer program that is being executed)。
广义定义
进程是一个具有一定独立功能的程序关于某个数据集合的一次运行活动。它是操作系统动态执行的基本单元,在传统的操作系统中,进程既是基本的分配单元,也是基本的执行单元。
举例:
比如py1文件中有个变量a=1,py2文件中有个变量a=2,他们两个会冲突吗?不会的,是不是,因为两个文件运行起来后是两个进程,操作系统让他们在内存上隔离开,对吧。
进程的特征
动态性:进程的实质是程序在多道程序系统中的一次执行过程,进程是动态产生,动态消亡的。
并发性:任何进程都可以同其他进程一起并发执行
独立性:进程是一个能独立运行的基本单位,同时也是系统分配资源和调度的独立单位;
异步性:由于进程间的相互制约,使进程具有执行的间断性,即进程按各自独立的、不可预知的速度向前推进
结构特征:进程由程序、数据和进程控制块三部分组成。
多个不同的进程可以包含相同的程序:一个程序在不同的数据集里就构成不同的进程,能得到不同的结果;但是执行过程中,程序不能发生改变。
并发性:任何进程都可以同其他进程一起并发执行
独立性:进程是一个能独立运行的基本单位,同时也是系统分配资源和调度的独立单位;
异步性:由于进程间的相互制约,使进程具有执行的间断性,即进程按各自独立的、不可预知的速度向前推进
结构特征:进程由程序、数据和进程控制块三部分组成。
多个不同的进程可以包含相同的程序:一个程序在不同的数据集里就构成不同的进程,能得到不同的结果;但是执行过程中,程序不能发生改变。
进程 与 程序 的区别
程序是指令和数据的有序集合,其本身没有任何运行的含义,是一个静态的概念。
而进程是程序在处理机上的一次执行过程,它是一个动态的概念。
程序可以作为一种软件资料长期存在,而进程是有一定生命期的。
程序是永久的,进程是暂时的。
举例:就像qq一样,qq是我们安装在自己电脑上的客户端程序,其实就是一堆的代码文件,我们不运行qq,那么他就是一堆代码程序,当我们运行qq的时候,这些代码运行起来,就成为一个进程了。
而进程是程序在处理机上的一次执行过程,它是一个动态的概念。
程序可以作为一种软件资料长期存在,而进程是有一定生命期的。
程序是永久的,进程是暂时的。
举例:就像qq一样,qq是我们安装在自己电脑上的客户端程序,其实就是一堆的代码文件,我们不运行qq,那么他就是一堆代码程序,当我们运行qq的时候,这些代码运行起来,就成为一个进程了。
注意:
同一个程序执行两次,就会在操作系统中出现两个进程,所以我们可以同时运行一个软件,分别做不同的事情也不会混乱。比如打开暴风影音,虽然都是同一个软件,但是一个可以播放苍井空,一个可以播放小泽玛利亚。
进程调度
先来先服务调度算法
短作业优先调度算法
时间片轮转法
多级反馈队列
并发与并行
并发
并发:是伪并行,即看起来是同时运行。单个cpu+多道技术就可以实现并发,(并行也属于并发)
单cpu,多进程并发举例
你是一个cpu,你同时谈了三个女朋友,每一个都可以是一个恋爱任务,你被这三个任务共享要玩出并发恋爱的效果,
应该是你先跟女友1去看电影,看了一会说:不好,我要拉肚子,然后跑去跟第二个女友吃饭,吃了一会说:那啥,我去趟洗手间,然后跑去跟女友3开了个房,然后在你的基友眼里,你就在和三个女友同时在一起玩。
应该是你先跟女友1去看电影,看了一会说:不好,我要拉肚子,然后跑去跟第二个女友吃饭,吃了一会说:那啥,我去趟洗手间,然后跑去跟女友3开了个房,然后在你的基友眼里,你就在和三个女友同时在一起玩。
并行
并行:同时运行,只有具备多个cpu才能实现并行
多个cpu、多进程并行举例
将多个cpu必须成高速公路上的多个车道,进程就好比每个车道上行驶的车辆,并行就是说,大家在自己的车道上行驶,会不影响,同时在开车。这就是并行
总结
通过进程之间的调度,也就是进程之间的切换,我们用户感知到的好像是两个视频文件同时在播放,或者音乐和游戏同时在进行,那就让我们来看一下什么叫做并发和并行
无论是并行还是并发,在用户看来都是'同时'运行的,不管是进程还是线程,都只是一个任务而已,真是干活的是cpu,cpu来做这些任务,而一个cpu同一时刻只能执行一个任务
无论是并行还是并发,在用户看来都是'同时'运行的,不管是进程还是线程,都只是一个任务而已,真是干活的是cpu,cpu来做这些任务,而一个cpu同一时刻只能执行一个任务
单核下,可以利用多道技术,多个核,每个核也都可以利用多道技术(多道技术是针对单核而言的)
有四个核,六个任务,这样同一时间有四个任务被执行,假设分别被分配给了cpu1,cpu2,cpu3,cpu4,
一旦任务1遇到I/O就被迫中断执行,此时任务5就拿到cpu1的时间片去执行,这就是单核下的多道技术
而一旦任务1的I/O结束了,操作系统会重新调用它(需知进程的调度、分配给哪个cpu运行,由操作系统说了算),可能被分配给四个cpu中的任意一个去执行 所有现代计算机经常会在同一时间做很多件事,一个用户的PC(无论是单cpu还是多cpu),都可以同时运行多个任务(一个任务可以理解为一个进程)。
有四个核,六个任务,这样同一时间有四个任务被执行,假设分别被分配给了cpu1,cpu2,cpu3,cpu4,
一旦任务1遇到I/O就被迫中断执行,此时任务5就拿到cpu1的时间片去执行,这就是单核下的多道技术
而一旦任务1的I/O结束了,操作系统会重新调用它(需知进程的调度、分配给哪个cpu运行,由操作系统说了算),可能被分配给四个cpu中的任意一个去执行 所有现代计算机经常会在同一时间做很多件事,一个用户的PC(无论是单cpu还是多cpu),都可以同时运行多个任务(一个任务可以理解为一个进程)。
同步\异步\阻塞\非阻塞(重点)
进程三状态
就绪状态
当进程已分配到除CPU以外的所有必要的资源,只要获得处理机便可立即执行,这时的进程状态称为就绪状态。
执行运行状态
执行/运行(Running)状态当进程已获得处理机,其程序正在处理机上执行,此时的进程状态称为执行状态。
阻塞状态
阻塞(Blocked)状态正在执行的进程,由于等待某个事件发生而无法执行时,便放弃处理机而处于阻塞状态。引起进程阻塞的事件可有多种,例如,等待I/O完成、申请缓冲区不能满足、等待信件(信号)等。
事件请求:input、sleep、文件输入输出、recv、accept等
事件发生:sleep、input等完成了
时间片到了之后有回到就绪状态,这三个状态不断的在转换。
事件请求:input、sleep、文件输入输出、recv、accept等
事件发生:sleep、input等完成了
时间片到了之后有回到就绪状态,这三个状态不断的在转换。
同步
所谓同步就是一个任务的完成需要依赖另外一个任务时,只有等待被依赖的任务完成后,依赖的任务才能算完成,这是一种可靠的任务序列。要成功都成功,失败都失败,两个任务的状态可以保持一致。其实就是一个程序结束才执行另外一个程序,串行的,不一定两个程序就有依赖关系。
异步
所谓异步是不需要等待被依赖的任务完成,只是通知被依赖的任务要完成什么工作,依赖的任务也立即执行,只要自己完成了整个任务就算完成了。至于被依赖的任务最终是否真正完成,依赖它的任务无法确定,所以它是不可靠的任务序列。
同步异步举例
比如我们去楼下的老家肉饼吃饭,饭点好了,取餐的时候发生了一些同步异步的事情。
同步:我们都站在队里等着取餐,前面有个人点了一份肉饼,后厨做了很久,但是由于同步机制,我们还是要站在队里等着前面那个人的肉饼做好取走,我们才往前走一步。
异步:我们点完餐之后,点餐员给了我们一个取餐号码,跟你说,你不用在这里排队等着,去找个地方坐着玩手机去吧,等饭做好了,我叫你。这种机制(等待别人通知)就是异步等待消息通知。在异步消息处理中,等待消息通知者(在这个例子中等着取餐的你)往往注册一个回调机制,在所等待的事件被触发时由触发机制(点餐员)通过某种机制(喊号,‘250号你的包子好了‘)找到等待该事件的人。
同步:我们都站在队里等着取餐,前面有个人点了一份肉饼,后厨做了很久,但是由于同步机制,我们还是要站在队里等着前面那个人的肉饼做好取走,我们才往前走一步。
异步:我们点完餐之后,点餐员给了我们一个取餐号码,跟你说,你不用在这里排队等着,去找个地方坐着玩手机去吧,等饭做好了,我叫你。这种机制(等待别人通知)就是异步等待消息通知。在异步消息处理中,等待消息通知者(在这个例子中等着取餐的你)往往注册一个回调机制,在所等待的事件被触发时由触发机制(点餐员)通过某种机制(喊号,‘250号你的包子好了‘)找到等待该事件的人。
阻塞 非阻塞
阻塞和非阻塞这两个概念与程序(线程)等待消息通知(无所谓同步或者异步)时的状态有关。也就是说阻塞与非阻塞主要是程序(线程)等待消息通知时的状态角度来说的
继续上面的那个例子,不论是排队还是使用号码等待通知,如果在这个等待的过程中,等待者除了等待消息通知之外不能做其它的事情,那么该机制就是阻塞的,表现在程序中,也就是该程序一直阻塞在该函数调用处不能继续往下执行。
相反,有的人喜欢在等待取餐的时候一边打游戏一边等待,这样的状态就是非阻塞的,因为他(等待者)没有阻塞在这个消息通知上,而是一边做自己的事情一边等待。阻塞的方法:input、time.sleep,socket中的recv、accept等等。
相反,有的人喜欢在等待取餐的时候一边打游戏一边等待,这样的状态就是非阻塞的,因为他(等待者)没有阻塞在这个消息通知上,而是一边做自己的事情一边等待。阻塞的方法:input、time.sleep,socket中的recv、accept等等。
.同步/异步 与 阻塞和非阻塞
同步阻塞形式
效率最低。拿上面的例子来说,就是你专心排队,什么别的事都不做。
异步阻塞形式
如果在排队取餐的人采用的是异步的方式去等待消息被触发(通知),也就是领了一张小纸条,假如在这段时间里他不能做其它的事情,就在那坐着等着,不能玩游戏等,那么很显然,这个人被阻塞在了这个等待的操作上面;
异步操作是可以被阻塞住的,只不过它不是在处理消息时阻塞,而是在等待消息通知时被阻塞。
同步非阻塞形式
实际上是效率低下的。
想象一下你一边打着电话一边还需要抬头看到底队伍排到你了没有,如果把打电话和观察排队的位置看成是程序的两个操作的话,这个程序需要在这两种不同的行为之间来回的切换,效率可想而知是低下的。
想象一下你一边打着电话一边还需要抬头看到底队伍排到你了没有,如果把打电话和观察排队的位置看成是程序的两个操作的话,这个程序需要在这两种不同的行为之间来回的切换,效率可想而知是低下的。
异步非阻塞形式
效率更高,
因为打电话是你(等待者)的事情,而通知你则是柜台(消息触发机制)的事情,程序没有在两种不同的操作中来回切换。
比如说,这个人突然发觉自己烟瘾犯了,需要出去抽根烟,于是他告诉点餐员说,排到我这个号码的时候麻烦到外面通知我一下,那么他就没有被阻塞在这个等待的操作上面,自然这个就是异步+非阻塞的方式了。
很多人会把同步和阻塞混淆,是因为很多时候同步操作会以阻塞的形式表现出来,同样的,很多人也会把异步和非阻塞混淆,因为异步操作一般都不会在真正的IO操作处被阻塞。
因为打电话是你(等待者)的事情,而通知你则是柜台(消息触发机制)的事情,程序没有在两种不同的操作中来回切换。
比如说,这个人突然发觉自己烟瘾犯了,需要出去抽根烟,于是他告诉点餐员说,排到我这个号码的时候麻烦到外面通知我一下,那么他就没有被阻塞在这个等待的操作上面,自然这个就是异步+非阻塞的方式了。
很多人会把同步和阻塞混淆,是因为很多时候同步操作会以阻塞的形式表现出来,同样的,很多人也会把异步和非阻塞混淆,因为异步操作一般都不会在真正的IO操作处被阻塞。
进程的创建、结束与并发的实现
进程的创建
凡是硬件,都需要有操作系统去管理,只要有操作系统,就有进程的概念,就需要有创建进程的方式,一些操作系统只为一个应用程序设计,比如微波炉中的控制器,一旦启动微波炉,所有的进程都已经存在。
第一种创建进程的方式代码实现()
import time
import os
from multiprocessing import Process
def func1():
time.sleep(2)
print('我是func1')
def func2():
time.sleep(3)
print('我是func2')
print('子进程的pid',os.getpid()) #查看当前进程的id
print('子进程的父进程>>>',os.getppid()) #查看父进程的id
if __name__ == '__main__':
print('主进程的pid',os.getpid())
print('当前主进程的父进程>>>',os.getppid())
start_time = time.time()
p = Process(target=func2,)
p.start() #告诉操作系统,我这边准备好了,你帮我创建并执行这个进程
func1()
end_time = time.time()
dif_time = end_time - start_time
print('>>>>',dif_time)
p.join()
# func1()
# func2()
import os
from multiprocessing import Process
def func1():
time.sleep(2)
print('我是func1')
def func2():
time.sleep(3)
print('我是func2')
print('子进程的pid',os.getpid()) #查看当前进程的id
print('子进程的父进程>>>',os.getppid()) #查看父进程的id
if __name__ == '__main__':
print('主进程的pid',os.getpid())
print('当前主进程的父进程>>>',os.getppid())
start_time = time.time()
p = Process(target=func2,)
p.start() #告诉操作系统,我这边准备好了,你帮我创建并执行这个进程
func1()
end_time = time.time()
dif_time = end_time - start_time
print('>>>>',dif_time)
p.join()
# func1()
# func2()
第二种创建进程的方式代码实现()
#第二种创建方式
class MyProess(Process):
def __init__(self,n):
super().__init__()
self.n = n
def run1(self):
time.sleep(2)
print(self.n)
# def run(self):
# time.sleep(2)
# print(self.n)
# print('aaaaa')
if __name__ == '__main__':
p = MyProess(10) #
p.start() #
print('主进程结束')
class MyProess(Process):
def __init__(self,n):
super().__init__()
self.n = n
def run1(self):
time.sleep(2)
print(self.n)
# def run(self):
# time.sleep(2)
# print(self.n)
# print('aaaaa')
if __name__ == '__main__':
p = MyProess(10) #
p.start() #
print('主进程结束')
进程的结束
1. 正常退出(自愿,如用户点击交互式页面的叉号,或程序执行完毕调用发起系统调用正常退出,在linux中用exit,在windows中用ExitProcess)
2. 出错退出(自愿,python a.py中a.py不存在)
3. 严重错误(非自愿,执行非法指令,如引用不存在的内存,1/0等,可以捕捉异常,try...except...)
4. 被其他进程杀死(非自愿,如kill -9)
2. 出错退出(自愿,python a.py中a.py不存在)
3. 严重错误(非自愿,执行非法指令,如引用不存在的内存,1/0等,可以捕捉异常,try...except...)
4. 被其他进程杀死(非自愿,如kill -9)
进程并发的实现
进程并发的实现在于,硬件中断一个正在运行的进程,把此时进程运行的所有状态保存下来,为此,操作系统维护一张表格,即进程表(process table),每个进程占用一个进程表项(这些表项也称为进程控制块)
该表存放了进程状态的重要信息:程序计数器、堆栈指针、内存分配状况、所有打开文件的状态、帐号和调度信息,以及其他在进程由运行态转为就绪态或阻塞态时,必须保存的信息,从而保证该进程在再次启动时,就像从未被中断过一样。
该表存放了进程状态的重要信息:程序计数器、堆栈指针、内存分配状况、所有打开文件的状态、帐号和调度信息,以及其他在进程由运行态转为就绪态或阻塞态时,必须保存的信息,从而保证该进程在再次启动时,就像从未被中断过一样。
python中进程的应用实战
multiprocess模块
仔细说来,multiprocess不是一个模块而是python中一个操作、管理进程的包。 将这部分大致分为四个部分:创建进程部分,进程同步部分,进程池部分,进程之间数据共享。重点强调:进程没有任何共享状态,进程修改的数据,改动仅限于该进程内,但是通过一些特殊的方法,可以实现进程之间数据的共享。
multiprocess模块 下的
process模块
process模块是一个创建进程的模块,借助这个模块,就可以完成进程的创建。
Process([group [, target [, name [, args [, kwargs]]]]]),由该类实例化得到的对象,表示一个子进程中的任务(尚未启动)
强调:
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数,是一个元组形式,必须有逗号
强调:
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数,是一个元组形式,必须有逗号
Process类中参数的介绍:
参数介绍:
1 group参数未使用,值始终为None
2 target表示调用对象,即子进程要执行的任务
3 args表示调用对象的位置参数元组,args=(1,2,'egon',)
4 kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}
5 name为子进程的名称
参数介绍:
1 group参数未使用,值始终为None
2 target表示调用对象,即子进程要执行的任务
3 args表示调用对象的位置参数元组,args=(1,2,'egon',)
4 kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}
5 name为子进程的名称
给要执行的函数传参数:
def func(x,y):
print(x)
time.sleep(1)
print(y)
if __name__ == '__main__':
p = Process(target=func,args=('姑娘','来玩啊!'))#这是func需要接收的参数的传送方式。
p.start()
print('父进程执行结束!')
#执行结果:
父进程执行结束!
姑娘
来玩啊!
函数传参
print(x)
time.sleep(1)
print(y)
if __name__ == '__main__':
p = Process(target=func,args=('姑娘','来玩啊!'))#这是func需要接收的参数的传送方式。
p.start()
print('父进程执行结束!')
#执行结果:
父进程执行结束!
姑娘
来玩啊!
函数传参
实例::#当前文件名称为test.py
# from multiprocessing import Process
#
# def func():
# print(12345)
#
# if __name__ == '__main__': #windows 下才需要写这个,这和系统创建进程的机制有关系,不用深究,记着windows下要写就好啦
# #首先我运行当前这个test.py文件,运行这个文件的程序,那么就产生了进程,这个进程我们称为主进程
#
# p = Process(target=func,) #将函数注册到一个进程中,p是一个进程对象,此时还没有启动进程,只是创建了一个进程对象。并且func是不加括号的,因为加上括号这个函数就直接运行了对吧。
# p.start() #告诉操作系统,给我开启一个进程,func这个函数就被我们新开的这个进程执行了,而这个进程是我主进程运行过程中创建出来的,所以称这个新创建的进程为主进程的子进程,而主进程又可以称为这个新进程的父进程。
#而这个子进程中执行的程序,相当于将现在这个test.py文件中的程序copy到一个你看不到的python文件中去执行了,就相当于当前这个文件,被另外一个py文件import过去并执行了。
#start并不是直接就去执行了,我们知道进程有三个状态,进程会进入进程的三个状态,就绪,(被调度,也就是时间片切换到它的时候)执行,阻塞,并且在这个三个状态之间不断的转换,等待cpu执行时间片到了。
# print('*' * 10) #这是主进程的程序,上面开启的子进程的程序是和主进程的程序同时运行的,我们称为异步
# from multiprocessing import Process
#
# def func():
# print(12345)
#
# if __name__ == '__main__': #windows 下才需要写这个,这和系统创建进程的机制有关系,不用深究,记着windows下要写就好啦
# #首先我运行当前这个test.py文件,运行这个文件的程序,那么就产生了进程,这个进程我们称为主进程
#
# p = Process(target=func,) #将函数注册到一个进程中,p是一个进程对象,此时还没有启动进程,只是创建了一个进程对象。并且func是不加括号的,因为加上括号这个函数就直接运行了对吧。
# p.start() #告诉操作系统,给我开启一个进程,func这个函数就被我们新开的这个进程执行了,而这个进程是我主进程运行过程中创建出来的,所以称这个新创建的进程为主进程的子进程,而主进程又可以称为这个新进程的父进程。
#而这个子进程中执行的程序,相当于将现在这个test.py文件中的程序copy到一个你看不到的python文件中去执行了,就相当于当前这个文件,被另外一个py文件import过去并执行了。
#start并不是直接就去执行了,我们知道进程有三个状态,进程会进入进程的三个状态,就绪,(被调度,也就是时间片切换到它的时候)执行,阻塞,并且在这个三个状态之间不断的转换,等待cpu执行时间片到了。
# print('*' * 10) #这是主进程的程序,上面开启的子进程的程序是和主进程的程序同时运行的,我们称为异步
通过主进程创建的子进程是异步执行的,那么我们就验证一下,并且看一下子进程和主进程(也就是父进程)的ID号(讲一下pid和ppid,使用pycharm举例),
来看看是否是父子关系。
来看看是否是父子关系。
import time
import os
#os.getpid() 获取自己进程的ID号
#os.getppid() 获取自己进程的父进程的ID号
from multiprocessing import Process
def func():
print('aaaa')
time.sleep(1)
print('子进程>>',os.getpid())
print('该子进程的父进程>>',os.getppid())
print(12345)
if __name__ == '__main__':
#首先我运行当前这个文件,运行的这个文件的程序,那么就产生了主进程
p = Process(target=func,)
p.start()
print('*' * 10)
print('父进程>>',os.getpid())
print('父进程的父进程>>',os.getppid())
#加上time和进程号给大家看一看结果:
#********** 首先打印出来了出进程的程序,然后打印的是子进程的,也就是子进程是异步执行的,相当于主进程和子进程同时运行着,如果是同步的话,我们先执行的是func(),然后再打印主进程最后的10个*号。
#父进程>> 3308
#父进程的父进程>> 5916 #我运行的test.py文件的父进程号,它是pycharm的进程号,看下面的截图
#aaaa
#子进程>> 4536
#该子进程的父进程>> 3308 #是我主进程的ID号,说明主进程为它的父进程
#12345
子进程与主进程
import os
#os.getpid() 获取自己进程的ID号
#os.getppid() 获取自己进程的父进程的ID号
from multiprocessing import Process
def func():
print('aaaa')
time.sleep(1)
print('子进程>>',os.getpid())
print('该子进程的父进程>>',os.getppid())
print(12345)
if __name__ == '__main__':
#首先我运行当前这个文件,运行的这个文件的程序,那么就产生了主进程
p = Process(target=func,)
p.start()
print('*' * 10)
print('父进程>>',os.getpid())
print('父进程的父进程>>',os.getppid())
#加上time和进程号给大家看一看结果:
#********** 首先打印出来了出进程的程序,然后打印的是子进程的,也就是子进程是异步执行的,相当于主进程和子进程同时运行着,如果是同步的话,我们先执行的是func(),然后再打印主进程最后的10个*号。
#父进程>> 3308
#父进程的父进程>> 5916 #我运行的test.py文件的父进程号,它是pycharm的进程号,看下面的截图
#aaaa
#子进程>> 4536
#该子进程的父进程>> 3308 #是我主进程的ID号,说明主进程为它的父进程
#12345
子进程与主进程
Process类中各方法的介绍:
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.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开启的进程
join 方法
join方法的例子:
让主进程加上join的地方等待(也就是阻塞住),等待子进程执行完之后,再继续往下执行我的主进程,好多时候,我们主进程需要子进程的执行结果,所以必须要等待。join感觉就像是将子进程和主进程拼接起来一样,将异步改为同步执行。
让主进程加上join的地方等待(也就是阻塞住),等待子进程执行完之后,再继续往下执行我的主进程,好多时候,我们主进程需要子进程的执行结果,所以必须要等待。join感觉就像是将子进程和主进程拼接起来一样,将异步改为同步执行。
def func(x,y):
print(x)
time.sleep(1)
print(y)
if __name__ == '__main__':
p = Process(target=func,args=('姑娘','来玩啊!'))
p.start()
print('我这里是异步的啊!') #这里相对于子进程还是异步的
p.join() #只有在join的地方才会阻塞住,将子进程和主进程之间的异步改为同步
print('父进程执行结束!')
#打印结果:
我这里是异步的啊!
姑娘
来玩啊!
父进程执行结束!
join方法的使用
print(x)
time.sleep(1)
print(y)
if __name__ == '__main__':
p = Process(target=func,args=('姑娘','来玩啊!'))
p.start()
print('我这里是异步的啊!') #这里相对于子进程还是异步的
p.join() #只有在join的地方才会阻塞住,将子进程和主进程之间的异步改为同步
print('父进程执行结束!')
#打印结果:
我这里是异步的啊!
姑娘
来玩啊!
父进程执行结束!
join方法的使用
Process类中自带封装的各属性的介绍
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字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性,这类连接只有在具有相同的身份验证键时才能成功(了解即可)
2 p.name:进程的名称
3 p.pid:进程的pid
4 p.exitcode:进程在运行时为None、如果为–N,表示被信号N结束(了解即可)
5 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性,这类连接只有在具有相同的身份验证键时才能成功(了解即可)
Process对象的其他方法或属性(简单了解一下就可以啦)
terminate,is_alive
#进程对象的其他方法一:terminate,is_alive
from multiprocessing import Process
import time
import random
class Piao(Process):
def __init__(self,name):
self.name=name
super().__init__()
def run(self):
print('%s is 打飞机' %self.name)
# s = input('???') #别忘了再pycharm下子进程中不能input输入,会报错EOFError: EOF when reading a line,因为子进程中没有像我们主进程这样的在pycharm下的控制台可以输入东西的地方
time.sleep(2)
print('%s is 打飞机结束' %self.name)
if __name__ == '__main__':
p1=Piao('太白')
p1.start()
time.sleep(5)
p1.terminate()#关闭进程,不会立即关闭,有个等着操作系统去关闭这个进程的时间,所以is_alive立刻查看的结果可能还是存活,但是稍微等一会,就被关掉了
print(p1.is_alive()) #结果为True
print('等会。。。。')
time.sleep(1)
print(p1.is_alive()) #结果为False
terminate和is_alive
from multiprocessing import Process
import time
import random
class Piao(Process):
def __init__(self,name):
self.name=name
super().__init__()
def run(self):
print('%s is 打飞机' %self.name)
# s = input('???') #别忘了再pycharm下子进程中不能input输入,会报错EOFError: EOF when reading a line,因为子进程中没有像我们主进程这样的在pycharm下的控制台可以输入东西的地方
time.sleep(2)
print('%s is 打飞机结束' %self.name)
if __name__ == '__main__':
p1=Piao('太白')
p1.start()
time.sleep(5)
p1.terminate()#关闭进程,不会立即关闭,有个等着操作系统去关闭这个进程的时间,所以is_alive立刻查看的结果可能还是存活,但是稍微等一会,就被关掉了
print(p1.is_alive()) #结果为True
print('等会。。。。')
time.sleep(1)
print(p1.is_alive()) #结果为False
terminate和is_alive
name 与 id
from multiprocessing import Process
import time
import random
class Piao(Process):
def __init__(self,name):
# self.name=name
# super().__init__() #Process的__init__方法会执行self.name=Piao-1,
# #所以加到这里,会覆盖我们的self.name=name
#为我们开启的进程设置名字的做法
super().__init__()
self.name=name
def run(self):
print('%s is piaoing' %self.name)
time.sleep(random.randrange(1,3))
print('%s is piao end' %self.name)
p=Piao('egon')
p.start()
print('开始')
print(p.pid) #查看pid
name与pid
import time
import random
class Piao(Process):
def __init__(self,name):
# self.name=name
# super().__init__() #Process的__init__方法会执行self.name=Piao-1,
# #所以加到这里,会覆盖我们的self.name=name
#为我们开启的进程设置名字的做法
super().__init__()
self.name=name
def run(self):
print('%s is piaoing' %self.name)
time.sleep(random.randrange(1,3))
print('%s is piao end' %self.name)
p=Piao('egon')
p.start()
print('开始')
print(p.pid) #查看pid
name与pid
僵尸进程与孤儿进程(简单了解 一下就可以啦)
一:僵尸进程(有害)
僵尸进程:一个进程使用fork创建子进程,如果子进程退出,而父进程并没有调用wait或waitpid获取子进程的状态信息,那么子进程的进程描述符仍然保存在系统中。这种进程称之为僵死进程。详解如下
我们知道在unix/linux中,正常情况下子进程是通过父进程创建的,子进程在创建新的进程。子进程的结束和父进程的运行是一个异步过程,即父进程永远无法预测子进程到底什么时候结束,如果子进程一结束就立刻回收其全部资源,那么在父进程内将无法获取子进程的状态信息。
因此,UNⅨ提供了一种机制可以保证父进程可以在任意时刻获取子进程结束时的状态信息:
1、在每个进程退出的时候,内核释放该进程所有的资源,包括打开的文件,占用的内存等。但是仍然为其保留一定的信息(包括进程号the process ID,退出状态the termination status of the process,运行时间the amount of CPU time taken by the process等)
2、直到父进程通过wait / waitpid来取时才释放. 但这样就导致了问题,如果进程不调用wait / waitpid的话,那么保留的那段信息就不会释放,其进程号就会一直被占用,但是系统所能使用的进程号是有限的,如果大量的产生僵死进程,将因为没有可用的进程号而导致系统不能产生新的进程. 此即为僵尸进程的危害,应当避免。
任何一个子进程(init除外)在exit()之后,并非马上就消失掉,而是留下一个称为僵尸进程(Zombie)的数据结构,等待父进程处理。这是每个子进程在结束时都要经过的阶段。如果子进程在exit()之后,父进程没有来得及处理,这时用ps命令就能看到子进程的状态是“Z”。如果父进程能及时 处理,可能用ps命令就来不及看到子进程的僵尸状态,但这并不等于子进程不经过僵尸状态。 如果父进程在子进程结束之前退出,则子进程将由init接管。init将会以父进程的身份对僵尸状态的子进程进行处理。
二:孤儿进程(无害)
孤儿进程:一个父进程退出,而它的一个或多个子进程还在运行,那么那些子进程将成为孤儿进程。孤儿进程将被init进程(进程号为1)所收养,并由init进程对它们完成状态收集工作。
孤儿进程是没有父进程的进程,孤儿进程这个重任就落到了init进程身上,init进程就好像是一个民政局,专门负责处理孤儿进程的善后工作。每当出现一个孤儿进程的时候,内核就把孤 儿进程的父进程设置为init,而init进程会循环地wait()它的已经退出的子进程。这样,当一个孤儿进程凄凉地结束了其生命周期的时候,init进程就会代表党和政府出面处理它的一切善后工作。因此孤儿进程并不会有什么危害。
我们来测试一下(创建完子进程后,主进程所在的这个脚本就退出了,当父进程先于子进程结束时,子进程会被init收养,成为孤儿进程,而非僵尸进程),文件内容
import os
import sys
import time
pid = os.getpid()
ppid = os.getppid()
print 'im father', 'pid', pid, 'ppid', ppid
pid = os.fork()
#执行pid=os.fork()则会生成一个子进程
#返回值pid有两种值:
# 如果返回的pid值为0,表示在子进程当中
# 如果返回的pid值>0,表示在父进程当中
if pid > 0:
print 'father died..'
sys.exit(0)
# 保证主线程退出完毕
time.sleep(1)
print 'im child', os.getpid(), os.getppid()
执行文件,输出结果:
im father pid 32515 ppid 32015
father died..
im child 32516 1
看,子进程已经被pid为1的init进程接收了,所以僵尸进程在这种情况下是不存在的,存在只有孤儿进程而已,孤儿进程声明周期结束自然会被init来销毁。
三:僵尸进程危害场景:
例如有个进程,它定期的产 生一个子进程,这个子进程需要做的事情很少,做完它该做的事情之后就退出了,因此这个子进程的生命周期很短,但是,父进程只管生成新的子进程,至于子进程 退出之后的事情,则一概不闻不问,这样,系统运行上一段时间之后,系统中就会存在很多的僵死进程,倘若用ps命令查看的话,就会看到很多状态为Z的进程。 严格地来说,僵死进程并不是问题的根源,罪魁祸首是产生出大量僵死进程的那个父进程。因此,当我们寻求如何消灭系统中大量的僵死进程时,答案就是把产生大 量僵死进程的那个元凶枪毙掉(也就是通过kill发送SIGTERM或者SIGKILL信号啦)。枪毙了元凶进程之后,它产生的僵死进程就变成了孤儿进 程,这些孤儿进程会被init进程接管,init进程会wait()这些孤儿进程,释放它们占用的系统进程表中的资源,这样,这些已经僵死的孤儿进程 就能瞑目而去了。
四:测试
#1、产生僵尸进程的程序test.py内容如下
#coding:utf-8
from multiprocessing import Process
import time,os
def run():
print('子',os.getpid())
if __name__ == '__main__':
p=Process(target=run)
p.start()
print('主',os.getpid())
time.sleep(1000)
#2、在unix或linux系统上执行
[root@vm172-31-0-19 ~]# python3 test.py &
[1] 18652
[root@vm172-31-0-19 ~]# 主 18652
子 18653
[root@vm172-31-0-19 ~]# ps aux |grep Z
USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND
root 18653 0.0 0.0 0 0 pts/0 Z 20:02 0:00 [python3] <defunct> #出现僵尸进程
root 18656 0.0 0.0 112648 952 pts/0 S+ 20:02 0:00 grep --color=auto Z
[root@vm172-31-0-19 ~]# top #执行top命令发现1zombie
top - 20:03:42 up 31 min, 3 users, load average: 0.01, 0.06, 0.12
Tasks: 93 total, 2 running, 90 sleeping, 0 stopped, 1 zombie
%Cpu(s): 0.0 us, 0.3 sy, 0.0 ni, 99.7 id, 0.0 wa, 0.0 hi, 0.0 si, 0.0 st
KiB Mem : 1016884 total, 97184 free, 70848 used, 848852 buff/cache
KiB Swap: 0 total, 0 free, 0 used. 782540 avail Mem
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
root 20 0 29788 1256 988 S 0.3 0.1 0:01.50 elfin
#3、
等待父进程正常结束后会调用wait/waitpid去回收僵尸进程
但如果父进程是一个死循环,永远不会结束,那么该僵尸进程就会一直存在,僵尸进程过多,就是有害的
解决方法一:杀死父进程
解决方法二:对开启的子进程应该记得使用join,join会回收僵尸进程
参考python2源码注释
class Process(object):
def join(self, timeout=None):
'''
Wait until child process terminates
'''
assert self._parent_pid == os.getpid(), 'can only join a child process'
assert self._popen is not None, 'can only join a started process'
res = self._popen.wait(timeout)
if res is not None:
_current_process._children.discard(self)
join方法中调用了wait,告诉系统释放僵尸进程。discard为从自己的children中剔除
解决方法三:http://blog.csdn.net/u010571844/article/details/50419798
僵尸进程与孤儿进程
一:僵尸进程(有害)
僵尸进程:一个进程使用fork创建子进程,如果子进程退出,而父进程并没有调用wait或waitpid获取子进程的状态信息,那么子进程的进程描述符仍然保存在系统中。这种进程称之为僵死进程。详解如下
我们知道在unix/linux中,正常情况下子进程是通过父进程创建的,子进程在创建新的进程。子进程的结束和父进程的运行是一个异步过程,即父进程永远无法预测子进程到底什么时候结束,如果子进程一结束就立刻回收其全部资源,那么在父进程内将无法获取子进程的状态信息。
因此,UNⅨ提供了一种机制可以保证父进程可以在任意时刻获取子进程结束时的状态信息:
1、在每个进程退出的时候,内核释放该进程所有的资源,包括打开的文件,占用的内存等。但是仍然为其保留一定的信息(包括进程号the process ID,退出状态the termination status of the process,运行时间the amount of CPU time taken by the process等)
2、直到父进程通过wait / waitpid来取时才释放. 但这样就导致了问题,如果进程不调用wait / waitpid的话,那么保留的那段信息就不会释放,其进程号就会一直被占用,但是系统所能使用的进程号是有限的,如果大量的产生僵死进程,将因为没有可用的进程号而导致系统不能产生新的进程. 此即为僵尸进程的危害,应当避免。
任何一个子进程(init除外)在exit()之后,并非马上就消失掉,而是留下一个称为僵尸进程(Zombie)的数据结构,等待父进程处理。这是每个子进程在结束时都要经过的阶段。如果子进程在exit()之后,父进程没有来得及处理,这时用ps命令就能看到子进程的状态是“Z”。如果父进程能及时 处理,可能用ps命令就来不及看到子进程的僵尸状态,但这并不等于子进程不经过僵尸状态。 如果父进程在子进程结束之前退出,则子进程将由init接管。init将会以父进程的身份对僵尸状态的子进程进行处理。
二:孤儿进程(无害)
孤儿进程:一个父进程退出,而它的一个或多个子进程还在运行,那么那些子进程将成为孤儿进程。孤儿进程将被init进程(进程号为1)所收养,并由init进程对它们完成状态收集工作。
孤儿进程是没有父进程的进程,孤儿进程这个重任就落到了init进程身上,init进程就好像是一个民政局,专门负责处理孤儿进程的善后工作。每当出现一个孤儿进程的时候,内核就把孤 儿进程的父进程设置为init,而init进程会循环地wait()它的已经退出的子进程。这样,当一个孤儿进程凄凉地结束了其生命周期的时候,init进程就会代表党和政府出面处理它的一切善后工作。因此孤儿进程并不会有什么危害。
我们来测试一下(创建完子进程后,主进程所在的这个脚本就退出了,当父进程先于子进程结束时,子进程会被init收养,成为孤儿进程,而非僵尸进程),文件内容
import os
import sys
import time
pid = os.getpid()
ppid = os.getppid()
print 'im father', 'pid', pid, 'ppid', ppid
pid = os.fork()
#执行pid=os.fork()则会生成一个子进程
#返回值pid有两种值:
# 如果返回的pid值为0,表示在子进程当中
# 如果返回的pid值>0,表示在父进程当中
if pid > 0:
print 'father died..'
sys.exit(0)
# 保证主线程退出完毕
time.sleep(1)
print 'im child', os.getpid(), os.getppid()
执行文件,输出结果:
im father pid 32515 ppid 32015
father died..
im child 32516 1
看,子进程已经被pid为1的init进程接收了,所以僵尸进程在这种情况下是不存在的,存在只有孤儿进程而已,孤儿进程声明周期结束自然会被init来销毁。
三:僵尸进程危害场景:
例如有个进程,它定期的产 生一个子进程,这个子进程需要做的事情很少,做完它该做的事情之后就退出了,因此这个子进程的生命周期很短,但是,父进程只管生成新的子进程,至于子进程 退出之后的事情,则一概不闻不问,这样,系统运行上一段时间之后,系统中就会存在很多的僵死进程,倘若用ps命令查看的话,就会看到很多状态为Z的进程。 严格地来说,僵死进程并不是问题的根源,罪魁祸首是产生出大量僵死进程的那个父进程。因此,当我们寻求如何消灭系统中大量的僵死进程时,答案就是把产生大 量僵死进程的那个元凶枪毙掉(也就是通过kill发送SIGTERM或者SIGKILL信号啦)。枪毙了元凶进程之后,它产生的僵死进程就变成了孤儿进 程,这些孤儿进程会被init进程接管,init进程会wait()这些孤儿进程,释放它们占用的系统进程表中的资源,这样,这些已经僵死的孤儿进程 就能瞑目而去了。
四:测试
#1、产生僵尸进程的程序test.py内容如下
#coding:utf-8
from multiprocessing import Process
import time,os
def run():
print('子',os.getpid())
if __name__ == '__main__':
p=Process(target=run)
p.start()
print('主',os.getpid())
time.sleep(1000)
#2、在unix或linux系统上执行
[root@vm172-31-0-19 ~]# python3 test.py &
[1] 18652
[root@vm172-31-0-19 ~]# 主 18652
子 18653
[root@vm172-31-0-19 ~]# ps aux |grep Z
USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND
root 18653 0.0 0.0 0 0 pts/0 Z 20:02 0:00 [python3] <defunct> #出现僵尸进程
root 18656 0.0 0.0 112648 952 pts/0 S+ 20:02 0:00 grep --color=auto Z
[root@vm172-31-0-19 ~]# top #执行top命令发现1zombie
top - 20:03:42 up 31 min, 3 users, load average: 0.01, 0.06, 0.12
Tasks: 93 total, 2 running, 90 sleeping, 0 stopped, 1 zombie
%Cpu(s): 0.0 us, 0.3 sy, 0.0 ni, 99.7 id, 0.0 wa, 0.0 hi, 0.0 si, 0.0 st
KiB Mem : 1016884 total, 97184 free, 70848 used, 848852 buff/cache
KiB Swap: 0 total, 0 free, 0 used. 782540 avail Mem
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
root 20 0 29788 1256 988 S 0.3 0.1 0:01.50 elfin
#3、
等待父进程正常结束后会调用wait/waitpid去回收僵尸进程
但如果父进程是一个死循环,永远不会结束,那么该僵尸进程就会一直存在,僵尸进程过多,就是有害的
解决方法一:杀死父进程
解决方法二:对开启的子进程应该记得使用join,join会回收僵尸进程
参考python2源码注释
class Process(object):
def join(self, timeout=None):
'''
Wait until child process terminates
'''
assert self._parent_pid == os.getpid(), 'can only join a child process'
assert self._popen is not None, 'can only join a started process'
res = self._popen.wait(timeout)
if res is not None:
_current_process._children.discard(self)
join方法中调用了wait,告诉系统释放僵尸进程。discard为从自己的children中剔除
解决方法三:http://blog.csdn.net/u010571844/article/details/50419798
僵尸进程与孤儿进程
注意:在windows中Process()必须放到# if __name__ == '__main__':下
验证进程之间是空间隔离的
#我们说进程之间的数据是隔离的,也就是数据不共享,看下面的验证
from multiprocessing import Process
n=100 #首先我定义了一个全局变量,在windows系统中应该把全局变量定义在if __name__ == '__main__'之上就可以了
def work():
global n
n=0
print('子进程内: ',n)
if __name__ == '__main__':
p=Process(target=work)
p.start()
p.join() #等待子进程执行完毕,如果数据共享的话,我子进程是不是通过global将n改为0了,但是你看打印结果,主进程在子进程执行结束之后,仍然是n=100,子进程n=0,说明子进程对n的修改没有在主进程中生效,说明什么?说明他们之间的数据是隔离的,互相不影响的
print('主进程内: ',n)
#看结果:
# 子进程内: 0
# 主进程内: 100
进程的内存空间是隔离的
from multiprocessing import Process
n=100 #首先我定义了一个全局变量,在windows系统中应该把全局变量定义在if __name__ == '__main__'之上就可以了
def work():
global n
n=0
print('子进程内: ',n)
if __name__ == '__main__':
p=Process(target=work)
p.start()
p.join() #等待子进程执行完毕,如果数据共享的话,我子进程是不是通过global将n改为0了,但是你看打印结果,主进程在子进程执行结束之后,仍然是n=100,子进程n=0,说明子进程对n的修改没有在主进程中生效,说明什么?说明他们之间的数据是隔离的,互相不影响的
print('主进程内: ',n)
#看结果:
# 子进程内: 0
# 主进程内: 100
进程的内存空间是隔离的
for循环开启多个进程呢?
并且我有个需求就是说,所有的子进程异步执行,然后所有的子进程全部执行完之后,我再执行主进程,怎么搞?看代码
#下面的注释按照编号去看,别忘啦!
import time
import os
from multiprocessing import Process
def func(x,y):
print(x)
# time.sleep(1) #进程切换:如果没有这个时间间隔,那么你会发现func执行结果是打印一个x然后一个y,再打印一个x一个y,不会出现打印多个x然后打印y的情况,因为两个打印距离太近了而且执行的也非常快,但是如果你这段程序运行慢的话,你就会发现进程之间的切换了。
print(y)
if __name__ == '__main__':
p_list= []
for i in range(10):
p = Process(target=func,args=('姑娘%s'%i,'来玩啊!'))
p_list.append(p)
p.start()
[ap.join() for ap in p_list] #4、这是解决办法,前提是我们的子进程全部都已经去执行了,那么我在一次给所有正在执行的子进程加上join,那么主进程就需要等着所有子进程执行结束才会继续执行自己的程序了,并且保障了所有子进程是异步执行的。
# p.join() #1、如果加到for循环里面,那么所有子进程包括父进程就全部变为同步了,因为for循环也是主进程的,循环第一次的时候,一个进程去执行了,然后这个进程就join住了,那么for循环就不会继续执行了,等着第一个子进程执行结束才会继续执行for循环去创建第二个子进程。
#2、如果我不想这样的,也就是我想所有的子进程是异步的,然后所有的子进程执行完了再执行主进程
#p.join() #3、如果这样写的话,多次运行之后,你会发现会出现主进程的程序比一些子进程先执行完,因为我们p.join()是对最后一个子进程进行了join,也就是说如果这最后一个子进程先于其他子进程执行完,那么主进程就会去执行,而此时如果还有一些子进程没有执行完,而主进程执行
#完了,那么就会先打印主进程的内容了,这个cpu调度进程的机制有关系,因为我们的电脑可能只有4个cpu,我的子进程加上住进程有11个,虽然我for循环是按顺序起进程的,但是操作系统一定会按照顺序给你执行你的进程吗,答案是不会的,操作系统会按照自己的算法来分配进
#程给cpu去执行,这里也解释了我们打印出来的子进程中的内容也是没有固定顺序的原因,因为打印结果也需要调用cpu,可以理解成进程在争抢cpu,如果同学你想问这是什么算法,这就要去研究操作系统啦。那我们的想所有子进程异步执行,然后再执行主进程的这个需求怎么解决啊
print('不要钱~~~~~~~~~~~~~~~~!')
代码
import time
import os
from multiprocessing import Process
def func(x,y):
print(x)
# time.sleep(1) #进程切换:如果没有这个时间间隔,那么你会发现func执行结果是打印一个x然后一个y,再打印一个x一个y,不会出现打印多个x然后打印y的情况,因为两个打印距离太近了而且执行的也非常快,但是如果你这段程序运行慢的话,你就会发现进程之间的切换了。
print(y)
if __name__ == '__main__':
p_list= []
for i in range(10):
p = Process(target=func,args=('姑娘%s'%i,'来玩啊!'))
p_list.append(p)
p.start()
[ap.join() for ap in p_list] #4、这是解决办法,前提是我们的子进程全部都已经去执行了,那么我在一次给所有正在执行的子进程加上join,那么主进程就需要等着所有子进程执行结束才会继续执行自己的程序了,并且保障了所有子进程是异步执行的。
# p.join() #1、如果加到for循环里面,那么所有子进程包括父进程就全部变为同步了,因为for循环也是主进程的,循环第一次的时候,一个进程去执行了,然后这个进程就join住了,那么for循环就不会继续执行了,等着第一个子进程执行结束才会继续执行for循环去创建第二个子进程。
#2、如果我不想这样的,也就是我想所有的子进程是异步的,然后所有的子进程执行完了再执行主进程
#p.join() #3、如果这样写的话,多次运行之后,你会发现会出现主进程的程序比一些子进程先执行完,因为我们p.join()是对最后一个子进程进行了join,也就是说如果这最后一个子进程先于其他子进程执行完,那么主进程就会去执行,而此时如果还有一些子进程没有执行完,而主进程执行
#完了,那么就会先打印主进程的内容了,这个cpu调度进程的机制有关系,因为我们的电脑可能只有4个cpu,我的子进程加上住进程有11个,虽然我for循环是按顺序起进程的,但是操作系统一定会按照顺序给你执行你的进程吗,答案是不会的,操作系统会按照自己的算法来分配进
#程给cpu去执行,这里也解释了我们打印出来的子进程中的内容也是没有固定顺序的原因,因为打印结果也需要调用cpu,可以理解成进程在争抢cpu,如果同学你想问这是什么算法,这就要去研究操作系统啦。那我们的想所有子进程异步执行,然后再执行主进程的这个需求怎么解决啊
print('不要钱~~~~~~~~~~~~~~~~!')
代码
模拟两个应用场景:1、同时对一个文件进行写操作 2、同时创建多个文件
import time
import os
import re
from multiprocessing import Process
#多进程同时对一个文件进行写操作
def func(x,y,i):
with open(x,'a',encoding='utf-8') as f:
print('当前进程%s拿到的文件的光标位置>>%s'%(os.getpid(),f.tell()))
f.write(y)
#多进程同时创建多个文件
# def func(x, y):
# with open(x, 'w', encoding='utf-8') as f:
# f.write(y)
if __name__ == '__main__':
p_list= []
for i in range(10):
p = Process(target=func,args=('can_do_girl_lists.txt','姑娘%s'%i,i))
# p = Process(target=func,args=('can_do_girl_info%s.txt'%i,'姑娘电话0000%s'%i))
p_list.append(p)
p.start()
[ap.join() for ap in p_list] #这就是个for循环,只不过用列表生成式的形式写的
with open('can_do_girl_lists.txt','r',encoding='utf-8') as f:
data = f.read()
all_num = re.findall('\d+',data) #打开文件,统计一下里面有多少个数据,每个数据都有个数字,所以re匹配一下就行了
print('>>>>>',all_num,'.....%s'%(len(all_num)))
#print([i in in os.walk(r'你的文件夹路径')])
print('不要钱~~~~~~~~~~~~~~~~!')
应用场景
import os
import re
from multiprocessing import Process
#多进程同时对一个文件进行写操作
def func(x,y,i):
with open(x,'a',encoding='utf-8') as f:
print('当前进程%s拿到的文件的光标位置>>%s'%(os.getpid(),f.tell()))
f.write(y)
#多进程同时创建多个文件
# def func(x, y):
# with open(x, 'w', encoding='utf-8') as f:
# f.write(y)
if __name__ == '__main__':
p_list= []
for i in range(10):
p = Process(target=func,args=('can_do_girl_lists.txt','姑娘%s'%i,i))
# p = Process(target=func,args=('can_do_girl_info%s.txt'%i,'姑娘电话0000%s'%i))
p_list.append(p)
p.start()
[ap.join() for ap in p_list] #这就是个for循环,只不过用列表生成式的形式写的
with open('can_do_girl_lists.txt','r',encoding='utf-8') as f:
data = f.read()
all_num = re.findall('\d+',data) #打开文件,统计一下里面有多少个数据,每个数据都有个数字,所以re匹配一下就行了
print('>>>>>',all_num,'.....%s'%(len(all_num)))
#print([i in in os.walk(r'你的文件夹路径')])
print('不要钱~~~~~~~~~~~~~~~~!')
应用场景
守护进程
子进程是不会随着主进程的结束而结束,子进程全部执行完之后,程序才结束,那么如果有一天我们的需求是我的主进程结束了,由我主进程创建的那些子进程必须跟着结束,怎么办?守护进程就来了!
主进程创建守护进程
其一:守护进程会在主进程代码执行结束后就终止
其二:守护进程内无法再开启子进程,否则抛出异常:AssertionError: daemonic processes are not allowed to have children
注意:进程之间是互相独立的,主进程代码运行结束,守护进程随即终止
主进程创建守护进程
其一:守护进程会在主进程代码执行结束后就终止
其二:守护进程内无法再开启子进程,否则抛出异常:AssertionError: daemonic processes are not allowed to have children
注意:进程之间是互相独立的,主进程代码运行结束,守护进程随即终止
代码演示
import os
import time
from multiprocessing import Process
class Myprocess(Process):
def __init__(self,person):
super().__init__()
self.person = person
def run(self):
print(os.getpid(),self.name)
print('%s正在和女主播聊天' %self.person)
time.sleep(3)
if __name__ == '__main__':
p=Myprocess('太白')
p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
p.start()
# time.sleep(1) # 在sleep时linux下查看进程id对应的进程ps -ef|grep id
print('主')
守护进程
import time
from multiprocessing import Process
class Myprocess(Process):
def __init__(self,person):
super().__init__()
self.person = person
def run(self):
print(os.getpid(),self.name)
print('%s正在和女主播聊天' %self.person)
time.sleep(3)
if __name__ == '__main__':
p=Myprocess('太白')
p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
p.start()
# time.sleep(1) # 在sleep时linux下查看进程id对应的进程ps -ef|grep id
print('主')
守护进程
进程同步(锁)
实现了程序的异步,让多个任务可以同时在几个进程中并发处理,他们之间的运行没有顺序,一旦开启也不受我们控制。
尽管并发编程让我们能更加充分的利用IO资源,但是也给我们带来了新的问题:进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,
或同一个打印终端,是没有问题的,而共享带来的是竞争,竞争带来的结果就是错乱,如何控制,就是加锁处理。
尽管并发编程让我们能更加充分的利用IO资源,但是也给我们带来了新的问题:进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,
或同一个打印终端,是没有问题的,而共享带来的是竞争,竞争带来的结果就是错乱,如何控制,就是加锁处理。
多进程抢占输出资源,导致打印混乱的示例
import os
import time
import random
from multiprocessing import Process
def work(n):
print('%s: %s is running' %(n,os.getpid()))
time.sleep(random.random())
print('%s:%s is done' %(n,os.getpid()))
if __name__ == '__main__':
for i in range(5):
p=Process(target=work,args=(i,))
p.start()
# 看结果:通过结果可以看出两个问题:问题一:每个进程中work函数的第一个打印就不是按照我们for循环的0-4的顺序来打印的
#问题二:我们发现,每个work进程中有两个打印,但是我们看到所有进程中第一个打印的顺序为0-2-1-4-3,但是第二个打印没有按照这个顺序,变成了2-1-0-3-4,说明我们一个进程中的程序的执行顺序都混乱了。
#问题的解决方法,第二个问题加锁来解决,第一个问题是没有办法解决的,因为进程开到了内核,有操作系统来决定进程的调度,我们自己控制不了
# 0: 9560 is running
# 2: 13824 is running
# 1: 7476 is running
# 4: 11296 is running
# 3: 14364 is running
# 2:13824 is done
# 1:7476 is done
# 0:9560 is done
# 3:14364 is done
# 4:11296 is done
多进程抢占输出资源,导致打印混乱的示例
import time
import random
from multiprocessing import Process
def work(n):
print('%s: %s is running' %(n,os.getpid()))
time.sleep(random.random())
print('%s:%s is done' %(n,os.getpid()))
if __name__ == '__main__':
for i in range(5):
p=Process(target=work,args=(i,))
p.start()
# 看结果:通过结果可以看出两个问题:问题一:每个进程中work函数的第一个打印就不是按照我们for循环的0-4的顺序来打印的
#问题二:我们发现,每个work进程中有两个打印,但是我们看到所有进程中第一个打印的顺序为0-2-1-4-3,但是第二个打印没有按照这个顺序,变成了2-1-0-3-4,说明我们一个进程中的程序的执行顺序都混乱了。
#问题的解决方法,第二个问题加锁来解决,第一个问题是没有办法解决的,因为进程开到了内核,有操作系统来决定进程的调度,我们自己控制不了
# 0: 9560 is running
# 2: 13824 is running
# 1: 7476 is running
# 4: 11296 is running
# 3: 14364 is running
# 2:13824 is done
# 1:7476 is done
# 0:9560 is done
# 3:14364 is done
# 4:11296 is done
多进程抢占输出资源,导致打印混乱的示例
示例: 加锁:由并发改成了串行,牺牲了运行效率,但避免了竞争
#由并发变成了串行,牺牲了运行效率,但避免了竞争
from multiprocessing import Process,Lock
import os,time
def work(n,lock):
#加锁,保证每次只有一个进程在执行锁里面的程序,这一段程序对于所有写上这个锁的进程,大家都变成了串行
lock.acquire()
print('%s: %s is running' %(n,os.getpid()))
time.sleep(1)
print('%s:%s is done' %(n,os.getpid()))
#解锁,解锁之后其他进程才能去执行自己的程序
lock.release()
if __name__ == '__main__':
lock=Lock()
for i in range(5):
p=Process(target=work,args=(i,lock))
p.start()
#打印结果:
# 2: 10968 is running
# 2:10968 is done
# 0: 7932 is running
# 0:7932 is done
# 4: 4404 is running
# 4:4404 is done
# 1: 12852 is running
# 1:12852 is done
# 3: 980 is running
# 3:980 is done
#结果分析:(自己去多次运行一下,看看结果,我拿出其中一个结果来看)通过结果我们可以看出,多进程刚开始去执行的时候,每次运行,首先打印出来哪个进程的程序是不固定的,但是我们解决了上面打印混乱示例代码的第二个问题,那就是同一个进程中的两次打印都是先完成的,然后才切换到下一个进程去,打印下一个进程中的两个打印结果,说明我们控制住了同一进程中的代码执行顺序,如果涉及到多个进程去操作同一个数据或者文件的时候,就不担心数据算错或者文件中的内容写入混乱了。
加锁:由并发改成了串行,牺牲了运行效率,但避免了竞争
from multiprocessing import Process,Lock
import os,time
def work(n,lock):
#加锁,保证每次只有一个进程在执行锁里面的程序,这一段程序对于所有写上这个锁的进程,大家都变成了串行
lock.acquire()
print('%s: %s is running' %(n,os.getpid()))
time.sleep(1)
print('%s:%s is done' %(n,os.getpid()))
#解锁,解锁之后其他进程才能去执行自己的程序
lock.release()
if __name__ == '__main__':
lock=Lock()
for i in range(5):
p=Process(target=work,args=(i,lock))
p.start()
#打印结果:
# 2: 10968 is running
# 2:10968 is done
# 0: 7932 is running
# 0:7932 is done
# 4: 4404 is running
# 4:4404 is done
# 1: 12852 is running
# 1:12852 is done
# 3: 980 is running
# 3:980 is done
#结果分析:(自己去多次运行一下,看看结果,我拿出其中一个结果来看)通过结果我们可以看出,多进程刚开始去执行的时候,每次运行,首先打印出来哪个进程的程序是不固定的,但是我们解决了上面打印混乱示例代码的第二个问题,那就是同一个进程中的两次打印都是先完成的,然后才切换到下一个进程去,打印下一个进程中的两个打印结果,说明我们控制住了同一进程中的代码执行顺序,如果涉及到多个进程去操作同一个数据或者文件的时候,就不担心数据算错或者文件中的内容写入混乱了。
加锁:由并发改成了串行,牺牲了运行效率,但避免了竞争
左边这种情况虽然使用加锁的形式实现了顺序的执行,但是程序又重新变成串行了,这样确实会浪费了时间,却保证了数据的安全。
接下来,我们以模拟抢票为例,来看看数据安全的重要性。
接下来,我们以模拟抢票为例,来看看数据安全的重要性。
并发运行,效率高,但是竞争同一个文件,导致数据混乱
#注意:首先在当前文件目录下创建一个名为db的文件
#文件db的内容为:{"count":1},只有这一行数据,并且注意,每次运行完了之后,文件中的1变成了0,你需要手动将0改为1,然后在去运行代码。
#注意一定要用双引号,不然json无法识别
#并发运行,效率高,但竞争写同一文件,数据写入错乱
from multiprocessing import Process,Lock
import time,json,random
#查看剩余票数
def search():
dic=json.load(open('db')) #打开文件,直接load文件中的内容,拿到文件中的包含剩余票数的字典
print('\033[43m剩余票数%s\033[0m' %dic['count'])
#抢票
def get():
dic=json.load(open('db'))
time.sleep(0.1) #模拟读数据的网络延迟,那么进程之间的切换,导致所有人拿到的字典都是{"count": 1},也就是每个人都拿到了这一票。
if dic['count'] >0:
dic['count']-=1
time.sleep(0.2) #模拟写数据的网络延迟
json.dump(dic,open('db','w'))
#最终结果导致,每个人显示都抢到了票,这就出现了问题~
print('\033[43m购票成功\033[0m')
def task():
search()
get()
if __name__ == '__main__':
for i in range(3): #模拟并发100个客户端抢票
p=Process(target=task)
p.start()
#看结果分析:由于网络延迟等原因使得进程切换,导致每个人都抢到了这最后一张票
# 剩余票数1
# 剩余票数1
# 剩余票数1
# 购票成功
# 购票成功
# 购票成功
并发运行,效率高,但是竞争同一个文件,导致数据混乱
#文件db的内容为:{"count":1},只有这一行数据,并且注意,每次运行完了之后,文件中的1变成了0,你需要手动将0改为1,然后在去运行代码。
#注意一定要用双引号,不然json无法识别
#并发运行,效率高,但竞争写同一文件,数据写入错乱
from multiprocessing import Process,Lock
import time,json,random
#查看剩余票数
def search():
dic=json.load(open('db')) #打开文件,直接load文件中的内容,拿到文件中的包含剩余票数的字典
print('\033[43m剩余票数%s\033[0m' %dic['count'])
#抢票
def get():
dic=json.load(open('db'))
time.sleep(0.1) #模拟读数据的网络延迟,那么进程之间的切换,导致所有人拿到的字典都是{"count": 1},也就是每个人都拿到了这一票。
if dic['count'] >0:
dic['count']-=1
time.sleep(0.2) #模拟写数据的网络延迟
json.dump(dic,open('db','w'))
#最终结果导致,每个人显示都抢到了票,这就出现了问题~
print('\033[43m购票成功\033[0m')
def task():
search()
get()
if __name__ == '__main__':
for i in range(3): #模拟并发100个客户端抢票
p=Process(target=task)
p.start()
#看结果分析:由于网络延迟等原因使得进程切换,导致每个人都抢到了这最后一张票
# 剩余票数1
# 剩余票数1
# 剩余票数1
# 购票成功
# 购票成功
# 购票成功
并发运行,效率高,但是竞争同一个文件,导致数据混乱
加锁:购票行为由并发变成了串行,牺牲了效率,但是保证了数据安全
#注意:首先在当前文件目录下创建一个名为db的文件
#文件db的内容为:{"count":1},只有这一行数据,并且注意,每次运行完了之后,文件中的1变成了0,你需要手动将0改为1,然后在去运行代码。
#注意一定要用双引号,不然json无法识别
#加锁保证数据安全,不出现混乱
from multiprocessing import Process,Lock
import time,json,random
#查看剩余票数
def search():
dic=json.load(open('db')) #打开文件,直接load文件中的内容,拿到文件中的包含剩余票数的字典
print('\033[43m剩余票数%s\033[0m' %dic['count'])
#抢票
def get():
dic=json.load(open('db'))
time.sleep(0.1) #模拟读数据的网络延迟,那么进程之间的切换,导致所有人拿到的字典都是{"count": 1},也就是每个人都拿到了这一票。
if dic['count'] >0:
dic['count']-=1
time.sleep(0.2) #模拟写数据的网络延迟
json.dump(dic,open('db','w'))
#最终结果导致,每个人显示都抢到了票,这就出现了问题~
print('\033[43m购票成功\033[0m')
else:
print('sorry,没票了亲!')
def task(lock):
search()
#因为抢票的时候是发生数据变化的时候,所有我们将锁加加到这里
lock.acquire()
get()
lock.release()
if __name__ == '__main__':
lock = Lock() #创建一个锁
for i in range(3): #模拟并发100个客户端抢票
p=Process(target=task,args=(lock,)) #将锁作为参数传给task函数
p.start()
#看结果分析:只有一个人抢到了票
# 剩余票数1
# 剩余票数1
# 剩余票数1
# 购票成功 #幸运的人儿
# sorry,没票了亲!
# sorry,没票了亲!
加锁:购票行为由并发变成了串行,牺牲了效率,但是保证了数据安全
#文件db的内容为:{"count":1},只有这一行数据,并且注意,每次运行完了之后,文件中的1变成了0,你需要手动将0改为1,然后在去运行代码。
#注意一定要用双引号,不然json无法识别
#加锁保证数据安全,不出现混乱
from multiprocessing import Process,Lock
import time,json,random
#查看剩余票数
def search():
dic=json.load(open('db')) #打开文件,直接load文件中的内容,拿到文件中的包含剩余票数的字典
print('\033[43m剩余票数%s\033[0m' %dic['count'])
#抢票
def get():
dic=json.load(open('db'))
time.sleep(0.1) #模拟读数据的网络延迟,那么进程之间的切换,导致所有人拿到的字典都是{"count": 1},也就是每个人都拿到了这一票。
if dic['count'] >0:
dic['count']-=1
time.sleep(0.2) #模拟写数据的网络延迟
json.dump(dic,open('db','w'))
#最终结果导致,每个人显示都抢到了票,这就出现了问题~
print('\033[43m购票成功\033[0m')
else:
print('sorry,没票了亲!')
def task(lock):
search()
#因为抢票的时候是发生数据变化的时候,所有我们将锁加加到这里
lock.acquire()
get()
lock.release()
if __name__ == '__main__':
lock = Lock() #创建一个锁
for i in range(3): #模拟并发100个客户端抢票
p=Process(target=task,args=(lock,)) #将锁作为参数传给task函数
p.start()
#看结果分析:只有一个人抢到了票
# 剩余票数1
# 剩余票数1
# 剩余票数1
# 购票成功 #幸运的人儿
# sorry,没票了亲!
# sorry,没票了亲!
加锁:购票行为由并发变成了串行,牺牲了效率,但是保证了数据安全
进程锁总结
#加锁可以保证多个进程修改同一块数据时,同一时间只能有一个任务可以进行修改,即串行的修改,没错,速度是慢了,但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信,但问题是:
1.效率低(共享数据基于文件,而文件是硬盘上的数据)
2.需要自己加锁处理
#因此我们最好找寻一种解决方案能够兼顾:1、效率高(多个进程共享一块内存的数据)2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制:队列和管道。
队列和管道都是将数据存放于内存中
队列又是基于(管道+锁)实现的,可以让我们从复杂的锁问题中解脱出来,
我们应该尽量避免使用共享数据,尽可能使用消息传递和队列,避免处理复杂的同步和锁问题,而且在进程数目增多时,往往可以获得更好的可获展性。
IPC通信机制(了解):IPC是intent-Process Communication的缩写,含义为进程间通信或者跨进程通信,是指两个进程之间进行数据交换的过程。IPC不是某个系统所独有的,任何一个操作系统都需要有相应的IPC机制,
比如Windows上可以通过剪贴板、管道和邮槽等来进行进程间通信,而Linux上可以通过命名共享内容、信号量等来进行进程间通信。Android它也有自己的进程间通信方式,Android建构在Linux基础上,继承了一
部分Linux的通信方式。
虽然可以用文件共享数据实现进程间通信,但问题是:
1.效率低(共享数据基于文件,而文件是硬盘上的数据)
2.需要自己加锁处理
#因此我们最好找寻一种解决方案能够兼顾:1、效率高(多个进程共享一块内存的数据)2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制:队列和管道。
队列和管道都是将数据存放于内存中
队列又是基于(管道+锁)实现的,可以让我们从复杂的锁问题中解脱出来,
我们应该尽量避免使用共享数据,尽可能使用消息传递和队列,避免处理复杂的同步和锁问题,而且在进程数目增多时,往往可以获得更好的可获展性。
IPC通信机制(了解):IPC是intent-Process Communication的缩写,含义为进程间通信或者跨进程通信,是指两个进程之间进行数据交换的过程。IPC不是某个系统所独有的,任何一个操作系统都需要有相应的IPC机制,
比如Windows上可以通过剪贴板、管道和邮槽等来进行进程间通信,而Linux上可以通过命名共享内容、信号量等来进行进程间通信。Android它也有自己的进程间通信方式,Android建构在Linux基础上,继承了一
部分Linux的通信方式。
队列(推荐使用)
进程彼此之间互相隔离,要实现进程间通信(IPC),multiprocessing模块支持两种形式:队列和管道,
这两种方式都是使用消息传递的。队列就像一个特殊的列表,
但是可以设置固定长度,并且从前面插入数据,从后面取出数据,先进先出。
这两种方式都是使用消息传递的。队列就像一个特殊的列表,
但是可以设置固定长度,并且从前面插入数据,从后面取出数据,先进先出。
Queue([maxsize]) 创建共享的进程队列。
参数 :maxsize是队列中允许的最大项数。如果省略此参数,则无大小限制。
底层队列使用管道和锁实现。
参数 :maxsize是队列中允许的最大项数。如果省略此参数,则无大小限制。
底层队列使用管道和锁实现。
queue的方法介绍
q = Queue([maxsize])
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数,则无大小限制。底层队列使用管道和锁定实现。另外,还需要运行支持线程以便队列中的数据传输到底层管道中。
Queue的实例q具有以下方法:
q.get( [ block [ ,timeout ] ] )
返回q中的一个项目。如果q为空,此方法将阻塞,直到队列中有项目可用为止。block用于控制阻塞行为,默认为True. 如果设置为False,将引发Queue.Empty异常(定义在Queue模块中)。timeout是可选超时时间,用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用,将引发Queue.Empty异常。
q.get_nowait( )
同q.get(False)方法。
q.put(item [, block [,timeout ] ] )
将item放入队列。如果队列已满,此方法将阻塞至有空间可用为止。block控制阻塞行为,默认为True。如果设置为False,将引发Queue.Empty异常(定义在Queue库模块中)。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。
q.qsize()
返回队列中目前项目的正确数量。此函数的结果并不可靠,因为在返回结果和在稍后程序中使用结果之间,队列中可能添加或删除了项目。在某些系统上,此方法可能引发NotImplementedError异常。
q.empty()
如果调用此方法时 q为空,返回True。如果其他进程或线程正在往队列中添加项目,结果是不可靠的。也就是说,在返回和使用结果之间,队列中可能已经加入新的项目。
q.full()
如果q已满,返回为True. 由于线程的存在,结果也可能是不可靠的(参考q.empty()方法)。。
方法介绍
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数,则无大小限制。底层队列使用管道和锁定实现。另外,还需要运行支持线程以便队列中的数据传输到底层管道中。
Queue的实例q具有以下方法:
q.get( [ block [ ,timeout ] ] )
返回q中的一个项目。如果q为空,此方法将阻塞,直到队列中有项目可用为止。block用于控制阻塞行为,默认为True. 如果设置为False,将引发Queue.Empty异常(定义在Queue模块中)。timeout是可选超时时间,用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用,将引发Queue.Empty异常。
q.get_nowait( )
同q.get(False)方法。
q.put(item [, block [,timeout ] ] )
将item放入队列。如果队列已满,此方法将阻塞至有空间可用为止。block控制阻塞行为,默认为True。如果设置为False,将引发Queue.Empty异常(定义在Queue库模块中)。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。
q.qsize()
返回队列中目前项目的正确数量。此函数的结果并不可靠,因为在返回结果和在稍后程序中使用结果之间,队列中可能添加或删除了项目。在某些系统上,此方法可能引发NotImplementedError异常。
q.empty()
如果调用此方法时 q为空,返回True。如果其他进程或线程正在往队列中添加项目,结果是不可靠的。也就是说,在返回和使用结果之间,队列中可能已经加入新的项目。
q.full()
如果q已满,返回为True. 由于线程的存在,结果也可能是不可靠的(参考q.empty()方法)。。
方法介绍
其他方法
q.close()
关闭队列,防止队列中加入更多数据。调用此方法时,后台线程将继续写入那些已入队列但尚未写入的数据,但将在此方法完成时马上关闭。如果q被垃圾收集,将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如,如果某个使用者正被阻塞在get()操作上,关闭生产者中的队列不会导致get()方法返回错误。
q.cancel_join_thread()
不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。
q.join_thread()
连接队列的后台线程。此方法用于在调用q.close()方法后,等待所有队列项被消耗。默认情况下,此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。
其他方法
关闭队列,防止队列中加入更多数据。调用此方法时,后台线程将继续写入那些已入队列但尚未写入的数据,但将在此方法完成时马上关闭。如果q被垃圾收集,将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如,如果某个使用者正被阻塞在get()操作上,关闭生产者中的队列不会导致get()方法返回错误。
q.cancel_join_thread()
不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。
q.join_thread()
连接队列的后台线程。此方法用于在调用q.close()方法后,等待所有队列项被消耗。默认情况下,此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。
其他方法
示例简单用法
from multiprocessing import Queue
q=Queue(3) #创建一个队列对象,队列长度为3
#put ,get ,put_nowait,get_nowait,full,empty
q.put(3) #往队列中添加数据
q.put(2)
q.put(1)
# q.put(4) # 如果队列已经满了,程序就会停在这里,等待数据被别人取走,再将数据放入队列。
# 如果队列中的数据一直不被取走,程序就会永远停在这里。
try:
q.put_nowait(4) # 可以使用put_nowait,如果队列满了不会阻塞,但是会因为队列满了而报错。
except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去,但是会丢掉这个消息。
print('队列已经满了')
# 因此,我们再放入数据之前,可以先看一下队列的状态,如果已经满了,就不继续put了。
print(q.full()) #查看是否满了,满了返回True,不满返回False
print(q.get()) #取出数据
print(q.get())
print(q.get())
# print(q.get()) # 同put方法一样,如果队列已经空了,那么继续取就会出现阻塞。
try:
q.get_nowait(3) # 可以使用get_nowait,如果队列满了不会阻塞,但是会因为没取到值而报错。
except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。
print('队列已经空了')
print(q.empty()) #空了
队列的简单用法
q=Queue(3) #创建一个队列对象,队列长度为3
#put ,get ,put_nowait,get_nowait,full,empty
q.put(3) #往队列中添加数据
q.put(2)
q.put(1)
# q.put(4) # 如果队列已经满了,程序就会停在这里,等待数据被别人取走,再将数据放入队列。
# 如果队列中的数据一直不被取走,程序就会永远停在这里。
try:
q.put_nowait(4) # 可以使用put_nowait,如果队列满了不会阻塞,但是会因为队列满了而报错。
except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去,但是会丢掉这个消息。
print('队列已经满了')
# 因此,我们再放入数据之前,可以先看一下队列的状态,如果已经满了,就不继续put了。
print(q.full()) #查看是否满了,满了返回True,不满返回False
print(q.get()) #取出数据
print(q.get())
print(q.get())
# print(q.get()) # 同put方法一样,如果队列已经空了,那么继续取就会出现阻塞。
try:
q.get_nowait(3) # 可以使用get_nowait,如果队列满了不会阻塞,但是会因为没取到值而报错。
except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。
print('队列已经空了')
print(q.empty()) #空了
队列的简单用法
生产者消费者模型
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯,而通过阻塞队列来进行通讯,所以生产者生产完数据之后不用等待消费者处理,直接扔给阻塞队列,消费者不找生产者要数据,而是直接从阻塞队列里取,阻塞队列就相当于一个缓冲区,平衡了生产者和消费者的处理能力,并且我可以根据生产速度和消费速度来均衡一下多少个生产者可以为多少个消费者提供足够的服务,就可以开多进程等等,而这些进程都是到阻塞队列或者说是缓冲区中去获取或者添加数据。
为什么要使用生产者和消费者模式
在线程世界里,生产者就是生产数据的线程,消费者就是消费数据的线程。
在多线程开发当中,如果生产者处理速度很快,而消费者处理速度很慢,那么生产者就必须等待消费者处理完,
才能继续生产数据。同样的道理,如果消费者的处理能力大于生产者,
那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
在多线程开发当中,如果生产者处理速度很快,而消费者处理速度很慢,那么生产者就必须等待消费者处理完,
才能继续生产数据。同样的道理,如果消费者的处理能力大于生产者,
那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
代码示例
from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
while True:
res=q.get()
time.sleep(random.randint(1,3))
print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))
def producer(q):
for i in range(10):
time.sleep(random.randint(1,3))
res='包子%s' %i
q.put(res)
print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))
if __name__ == '__main__':
q=Queue()
#生产者们:即厨师们
p1=Process(target=producer,args=(q,))
#消费者们:即吃货们
c1=Process(target=consumer,args=(q,))
#开始
p1.start()
c1.start()
print('主')
基于队列的生产者消费者模型
import time,random,os
def consumer(q):
while True:
res=q.get()
time.sleep(random.randint(1,3))
print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))
def producer(q):
for i in range(10):
time.sleep(random.randint(1,3))
res='包子%s' %i
q.put(res)
print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))
if __name__ == '__main__':
q=Queue()
#生产者们:即厨师们
p1=Process(target=producer,args=(q,))
#消费者们:即吃货们
c1=Process(target=consumer,args=(q,))
#开始
p1.start()
c1.start()
print('主')
基于队列的生产者消费者模型
总结
#生产者消费者模型总结
#程序中有两类角色
一类负责生产数据(生产者)
一类负责处理数据(消费者)
#引入生产者消费者模型为了解决的问题是:
平衡生产者与消费者之间的工作能力,从而提高程序整体处理数据的速度
#如何实现:
生产者<-->队列<——>消费者
#生产者消费者模型实现类程序的解耦和
生产者消费者模型总结
#程序中有两类角色
一类负责生产数据(生产者)
一类负责处理数据(消费者)
#引入生产者消费者模型为了解决的问题是:
平衡生产者与消费者之间的工作能力,从而提高程序整体处理数据的速度
#如何实现:
生产者<-->队列<——>消费者
#生产者消费者模型实现类程序的解耦和
生产者消费者模型总结
注意:
子进程生产者在生产完毕后发送结束信号None
主进程在生产者生产完毕后发送结束信号None
有多个消费者和生产者的时候需要发送多次结束信号
另外一种队列提供了这种机制
JoinableQueue([maxsize])
主进程在生产者生产完毕后发送结束信号None
有多个消费者和生产者的时候需要发送多次结束信号
另外一种队列提供了这种机制
JoinableQueue([maxsize])
JoinableQueue([maxsize])
#JoinableQueue([maxsize]):这就像是一个Queue对象,但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。
#参数介绍:
maxsize是队列中允许最大项数,省略则无大小限制。
#方法介绍:
JoinableQueue的实例p除了与Queue对象相同的方法之外还具有:
q.task_done():使用者使用此方法发出信号,表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量,将引发ValueError异常
q.join():生产者调用此方法进行阻塞,直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done()方法为止,也就是队列中的数据全部被get拿走了。
#参数介绍:
maxsize是队列中允许最大项数,省略则无大小限制。
#方法介绍:
JoinableQueue的实例p除了与Queue对象相同的方法之外还具有:
q.task_done():使用者使用此方法发出信号,表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量,将引发ValueError异常
q.join():生产者调用此方法进行阻塞,直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done()方法为止,也就是队列中的数据全部被get拿走了。
代码示例
from multiprocessing import Process,JoinableQueue
import time,random,os
def consumer(q):
while True:
res=q.get()
# time.sleep(random.randint(1,3))
time.sleep(random.random())
print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))
q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走并执行完了
def producer(name,q):
for i in range(10):
# time.sleep(random.randint(1,3))
time.sleep(random.random())
res='%s%s' %(name,i)
q.put(res)
print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))
print('%s生产结束'%name)
q.join() #生产完毕,使用此方法进行阻塞,直到队列中所有项目均被处理。
print('%s生产结束~~~~~~'%name)
if __name__ == '__main__':
q=JoinableQueue()
#生产者们:即厨师们
p1=Process(target=producer,args=('包子',q))
p2=Process(target=producer,args=('骨头',q))
p3=Process(target=producer,args=('泔水',q))
#消费者们:即吃货们
c1=Process(target=consumer,args=(q,))
c2=Process(target=consumer,args=(q,))
c1.daemon=True #如果不加守护,那么主进程结束不了,但是加了守护之后,必须确保生产者的内容生产完并且被处理完了,所有必须还要在主进程给生产者设置join,才能确保生产者生产的任务被执行完了,并且能够确保守护进程在所有任务执行完成之后才随着主进程的结束而结束。
c2.daemon=True
#开始
p_l=[p1,p2,p3,c1,c2]
for p in p_l:
p.start()
p1.join() #我要确保你的生产者进程结束了,生产者进程的结束标志着你生产的所有的人任务都已经被处理完了
p2.join()
p3.join()
print('主')
# 主进程等--->p1,p2,p3等---->c1,c2
# p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据
# 因而c1,c2也没有存在的价值了,不需要继续阻塞在进程中影响主进程了。应该随着主进程的结束而结束,所以设置成守护进程就可以了。
JoinableQueue队列实现生产者消费者模型
import time,random,os
def consumer(q):
while True:
res=q.get()
# time.sleep(random.randint(1,3))
time.sleep(random.random())
print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))
q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走并执行完了
def producer(name,q):
for i in range(10):
# time.sleep(random.randint(1,3))
time.sleep(random.random())
res='%s%s' %(name,i)
q.put(res)
print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))
print('%s生产结束'%name)
q.join() #生产完毕,使用此方法进行阻塞,直到队列中所有项目均被处理。
print('%s生产结束~~~~~~'%name)
if __name__ == '__main__':
q=JoinableQueue()
#生产者们:即厨师们
p1=Process(target=producer,args=('包子',q))
p2=Process(target=producer,args=('骨头',q))
p3=Process(target=producer,args=('泔水',q))
#消费者们:即吃货们
c1=Process(target=consumer,args=(q,))
c2=Process(target=consumer,args=(q,))
c1.daemon=True #如果不加守护,那么主进程结束不了,但是加了守护之后,必须确保生产者的内容生产完并且被处理完了,所有必须还要在主进程给生产者设置join,才能确保生产者生产的任务被执行完了,并且能够确保守护进程在所有任务执行完成之后才随着主进程的结束而结束。
c2.daemon=True
#开始
p_l=[p1,p2,p3,c1,c2]
for p in p_l:
p.start()
p1.join() #我要确保你的生产者进程结束了,生产者进程的结束标志着你生产的所有的人任务都已经被处理完了
p2.join()
p3.join()
print('主')
# 主进程等--->p1,p2,p3等---->c1,c2
# p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据
# 因而c1,c2也没有存在的价值了,不需要继续阻塞在进程中影响主进程了。应该随着主进程的结束而结束,所以设置成守护进程就可以了。
JoinableQueue队列实现生产者消费者模型
管道(了解)
进程间通信(IPC)方式二:管道(不推荐使用,了解即可),会导致数据不安全的情况出现,后面我们会说到为什么会带来数据 不安全的问题。
管道的使用
from multiprocessing import Process, Pipe
def f(conn):
conn.send("Hello 妹妹") #子进程发送了消息
conn.close()
if __name__ == '__main__':
parent_conn, child_conn = Pipe() #建立管道,拿到管道的两端,双工通信方式,两端都可以收发消息
p = Process(target=f, args=(child_conn,)) #将管道的一段给子进程
p.start() #开启子进程
print(parent_conn.recv()) #主进程接受了消息
p.join()
管道初使用
def f(conn):
conn.send("Hello 妹妹") #子进程发送了消息
conn.close()
if __name__ == '__main__':
parent_conn, child_conn = Pipe() #建立管道,拿到管道的两端,双工通信方式,两端都可以收发消息
p = Process(target=f, args=(child_conn,)) #将管道的一段给子进程
p.start() #开启子进程
print(parent_conn.recv()) #主进程接受了消息
p.join()
管道初使用
数据共享(了解)
Manager模块介绍
多进程共同去处理共享数据的时候,就和我们多进程同时去操作一个文件中的数据是一样的,不加锁就会出现错误的结果,进程不安全的,所以也需要加锁
from multiprocessing import Manager,Process,Lock
def work(d,lock):
with lock: #不加锁而操作共享的数据,肯定会出现数据错乱
d['count']-=1
if __name__ == '__main__':
lock=Lock()
with Manager() as m:
dic=m.dict({'count':100})
p_l=[]
for i in range(100):
p=Process(target=work,args=(dic,lock))
p_l.append(p)
p.start()
for p in p_l:
p.join()
print(dic)
Manager模块使用
def work(d,lock):
with lock: #不加锁而操作共享的数据,肯定会出现数据错乱
d['count']-=1
if __name__ == '__main__':
lock=Lock()
with Manager() as m:
dic=m.dict({'count':100})
p_l=[]
for i in range(100):
p=Process(target=work,args=(dic,lock))
p_l.append(p)
p.start()
for p in p_l:
p.join()
print(dic)
Manager模块使用
信号量(了解)
互斥锁同时只允许一个线程更改数据,而信号量Semaphore是同时允许一定数量的线程更改数据 。
假设商场里有4个迷你唱吧,所以同时可以进去4个人,如果来了第五个人就要在外面等待,等到有人出来才能再进去玩。
实现:
信号量同步基于内部计数器,每调用一次acquire(),计数器减1;每调用一次release(),计数器加1.当计数器为0时,acquire()调用被阻塞。这是迪科斯彻(Dijkstra)信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器这样的有限资源。
信号量与进程池的概念很像,但是要区分开,信号量涉及到加锁的概念
比如大保健:提前设定好,一个房间只有4个床(计数器现在为4),那么同时只能四个人进来,谁先来的谁先占一个床(acquire,计数器减1),4个床满了之后(计数器为0了),第五个人就要等着,等其中一个人出来(release,计数器加1),他就去占用那个床了。
假设商场里有4个迷你唱吧,所以同时可以进去4个人,如果来了第五个人就要在外面等待,等到有人出来才能再进去玩。
实现:
信号量同步基于内部计数器,每调用一次acquire(),计数器减1;每调用一次release(),计数器加1.当计数器为0时,acquire()调用被阻塞。这是迪科斯彻(Dijkstra)信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器这样的有限资源。
信号量与进程池的概念很像,但是要区分开,信号量涉及到加锁的概念
比如大保健:提前设定好,一个房间只有4个床(计数器现在为4),那么同时只能四个人进来,谁先来的谁先占一个床(acquire,计数器减1),4个床满了之后(计数器为0了),第五个人就要等着,等其中一个人出来(release,计数器加1),他就去占用那个床了。
from multiprocessing import Process,Semaphore
import time,random
def go_ktv(sem,user):
sem.acquire()
print('%s 占到一间ktv小屋' %user)
time.sleep(random.randint(0,3)) #模拟每个人在ktv中待的时间不同
sem.release()
if __name__ == '__main__':
sem=Semaphore(4)
p_l=[]
for i in range(13):
p=Process(target=go_ktv,args=(sem,'user%s' %i,))
p.start()
p_l.append(p)
for i in p_l:
i.join()
print('============》')
信号量使用
import time,random
def go_ktv(sem,user):
sem.acquire()
print('%s 占到一间ktv小屋' %user)
time.sleep(random.randint(0,3)) #模拟每个人在ktv中待的时间不同
sem.release()
if __name__ == '__main__':
sem=Semaphore(4)
p_l=[]
for i in range(13):
p=Process(target=go_ktv,args=(sem,'user%s' %i,))
p.start()
p_l.append(p)
for i in p_l:
i.join()
print('============》')
信号量使用
事件(了解)
python线程的事件用于主线程控制其他线程的执行,事件主要提供了三个方法 set、wait、clear。
事件处理的机制:全局定义了一个“Flag”,如果“Flag”值为 False,那么当程序执行 event.wait 方法时就会阻塞,如果“Flag”值为True,那么event.wait 方法时便不再阻塞。
clear:将“Flag”设置为False
set:将“Flag”设置为True
事件处理的机制:全局定义了一个“Flag”,如果“Flag”值为 False,那么当程序执行 event.wait 方法时就会阻塞,如果“Flag”值为True,那么event.wait 方法时便不再阻塞。
clear:将“Flag”设置为False
set:将“Flag”设置为True
from multiprocessing import Process,Semaphore,Event
import time,random
e = Event() #创建一个事件对象
print(e.is_set()) #is_set()查看一个事件的状态,默认为False,可通过set方法改为True
print('look here!')
# e.set() #将is_set()的状态改为True。
# print(e.is_set())#is_set()查看一个事件的状态,默认为False,可通过set方法改为Tr
# e.clear() #将is_set()的状态改为False
# print(e.is_set())#is_set()查看一个事件的状态,默认为False,可通过set方法改为Tr
e.wait() #根据is_set()的状态结果来决定是否在这阻塞住,is_set()=False那么就阻塞,is_set()=True就不阻塞
print('give me!!')
#set和clear 修改事件的状态 set-->True clear-->False
#is_set 用来查看一个事件的状态
#wait 依据事件的状态来决定是否阻塞 False-->阻塞 True-->不阻塞
事件方法的使用
import time,random
e = Event() #创建一个事件对象
print(e.is_set()) #is_set()查看一个事件的状态,默认为False,可通过set方法改为True
print('look here!')
# e.set() #将is_set()的状态改为True。
# print(e.is_set())#is_set()查看一个事件的状态,默认为False,可通过set方法改为Tr
# e.clear() #将is_set()的状态改为False
# print(e.is_set())#is_set()查看一个事件的状态,默认为False,可通过set方法改为Tr
e.wait() #根据is_set()的状态结果来决定是否在这阻塞住,is_set()=False那么就阻塞,is_set()=True就不阻塞
print('give me!!')
#set和clear 修改事件的状态 set-->True clear-->False
#is_set 用来查看一个事件的状态
#wait 依据事件的状态来决定是否阻塞 False-->阻塞 True-->不阻塞
事件方法的使用
小总结
总结一下,进程之间的通信:队列、管道、数据共享也算
讲的信号量和事件也相当于锁,也是全局的,所有进程都能拿到这些锁的状态,进程之间这些锁啊信号量啊事件啊等等的通信,其实底层还是socekt,只不过是基于文件的socket通信,而不是跟上面的数据共享啊空间共享啊之类的机制,我们之前学的是基于网络的socket通信,还记得socket的两个家族吗,一个文件的一个网络的,所以将来如果说这些锁之类的报错,可能你看到的就是类似于socket的错误,简单知道一下就可以啦~~~
工作中常用的是锁,信号量和事件不常用,但是信号量和事件面试的时候会问到,你能知道就行啦~~~
讲的信号量和事件也相当于锁,也是全局的,所有进程都能拿到这些锁的状态,进程之间这些锁啊信号量啊事件啊等等的通信,其实底层还是socekt,只不过是基于文件的socket通信,而不是跟上面的数据共享啊空间共享啊之类的机制,我们之前学的是基于网络的socket通信,还记得socket的两个家族吗,一个文件的一个网络的,所以将来如果说这些锁之类的报错,可能你看到的就是类似于socket的错误,简单知道一下就可以啦~~~
工作中常用的是锁,信号量和事件不常用,但是信号量和事件面试的时候会问到,你能知道就行啦~~~
进程池
为什么要有进程池?进程池的概念。
进程池的概念,定义一个池子,在里面放上固定数量的进程,有需求来了,就拿一个池中的进程来处理任务,等到处理完毕,进程并不关闭,而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行,池中的进程数量不够,任务就要等待之前的进程执行任务完毕归来,拿到空闲进程才能继续执行。也就是说,池中进程的数量是固定的,那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度,还节省了开闭进程的时间,也一定程度上能够实现并发效果
multiprocess.Poll模块
代码示例
import time
from multiprocessing import Process,Pool
def func(i):
num = 0
for j in range(5):
num += i
if __name__ == '__main__':
pool = Pool(4) #
p_list = []
start_time = time.time()
for i in range(500):
p = Process(target=func,args=(i,))
p_list.append(p)
p.start()
[pp.join() for pp in p_list]
end_time = time.time()
print(end_time - start_time)
s_time = time.time()
pool.map(func,range(500)) #map
e_time = time.time()
print(e_time - s_time)
from multiprocessing import Process,Pool
def func(i):
num = 0
for j in range(5):
num += i
if __name__ == '__main__':
pool = Pool(4) #
p_list = []
start_time = time.time()
for i in range(500):
p = Process(target=func,args=(i,))
p_list.append(p)
p.start()
[pp.join() for pp in p_list]
end_time = time.time()
print(end_time - start_time)
s_time = time.time()
pool.map(func,range(500)) #map
e_time = time.time()
print(e_time - s_time)
进程池同步方法
import time
from multiprocessing import Process,Pool
def func(i):
num = 0
for j in range(5):
num += i
if __name__ == '__main__':
pool = Pool(4) #
p_list = []
start_time = time.time()
for i in range(500):
p = Process(target=func,args=(i,))
p_list.append(p)
p.start()
[pp.join() for pp in p_list]
end_time = time.time()
print(end_time - start_time)
s_time = time.time()
pool.map(func,range(500)) #map
e_time = time.time()
print(e_time - s_time)
from multiprocessing import Process,Pool
def func(i):
num = 0
for j in range(5):
num += i
if __name__ == '__main__':
pool = Pool(4) #
p_list = []
start_time = time.time()
for i in range(500):
p = Process(target=func,args=(i,))
p_list.append(p)
p.start()
[pp.join() for pp in p_list]
end_time = time.time()
print(end_time - start_time)
s_time = time.time()
pool.map(func,range(500)) #map
e_time = time.time()
print(e_time - s_time)
进程池异步方法
import time
from multiprocessing import Process,Pool
def func(i):
num = 0
for j in range(5):
num += i
time.sleep(1)
# print('>>>>>',num)
return num
if __name__ == '__main__':
pool = Pool(4)
red_list = []
for i in range(10):
res = pool.apply_async(func,args=(i,)) #
# print(res)
red_list.append(res)
# print(res.get())
# pool.close() #不是关闭进程池,只是锁定
# pool.join() # 等待进程池中所有的任务执行完,但是无法确认是否所有的任务真的全部执行完了,前面要加close方法
for resss in red_list:
print(resss.get())
import time
from multiprocessing import Process,Pool
def func(i):
num = 0
for j in range(5):
num += i
time.sleep(1)
# print('>>>>>',num)
return num
if __name__ == '__main__':
pool = Pool(4)
red_list = []
for i in range(10):
res = pool.apply_async(func,args=(i,)) #
# print(res)
red_list.append(res)
# print(res.get())
# pool.close() #不是关闭进程池,只是锁定
# pool.join() # 等待进程池中所有的任务执行完,但是无法确认是否所有的任务真的全部执行完了,前面要加close方法
for resss in red_list:
print(resss.get())
进程池map 方法
进程池的简单应用及与进程池的效率对比
import time
from multiprocessing import Pool,Process
#针对range(100)这种参数的
# def func(n):
# for i in range(3):
# print(n + 1)
def func(n):
print(n)
# 结果:
# (1, 2)
# alex
def func2(n):
for i in range(3):
print(n - 1)
if __name__ == '__main__':
#1.进程池的模式
s1 = time.time() #我们计算一下开多进程和进程池的执行效率
poll = Pool(5) #创建含有5个进程的进程池
# poll.map(func,range(100)) #异步调用进程,开启100个任务,map自带join的功能
poll.map(func,[(1,2),'alex']) #异步调用进程,开启100个任务,map自带join的功能
# poll.map(func2,range(100)) #如果想让进程池完成不同的任务,可以直接这样搞
#map只限于接收一个可迭代的数据类型参数,列表啊,元祖啊等等,如果想做其他的参数之类的操作,需要用后面我们要学的方法。
# t1 = time.time() - s1
#
# #2.多进程的模式
# s2 = time.time()
# p_list = []
# for i in range(100):
# p = Process(target=func,args=(i,))
# p_list.append(p)
# p.start()
# [pp.join() for pp in p_list]
# t2 = time.time() - s2
#
# print('t1>>',t1) #结果:0.5146853923797607s 进程池的效率高
# print('t2>>',t2) #结果:12.092015027999878s
进程池的简单应用及与进程池的效率对比
from multiprocessing import Pool,Process
#针对range(100)这种参数的
# def func(n):
# for i in range(3):
# print(n + 1)
def func(n):
print(n)
# 结果:
# (1, 2)
# alex
def func2(n):
for i in range(3):
print(n - 1)
if __name__ == '__main__':
#1.进程池的模式
s1 = time.time() #我们计算一下开多进程和进程池的执行效率
poll = Pool(5) #创建含有5个进程的进程池
# poll.map(func,range(100)) #异步调用进程,开启100个任务,map自带join的功能
poll.map(func,[(1,2),'alex']) #异步调用进程,开启100个任务,map自带join的功能
# poll.map(func2,range(100)) #如果想让进程池完成不同的任务,可以直接这样搞
#map只限于接收一个可迭代的数据类型参数,列表啊,元祖啊等等,如果想做其他的参数之类的操作,需要用后面我们要学的方法。
# t1 = time.time() - s1
#
# #2.多进程的模式
# s2 = time.time()
# p_list = []
# for i in range(100):
# p = Process(target=func,args=(i,))
# p_list.append(p)
# p.start()
# [pp.join() for pp in p_list]
# t2 = time.time() - s2
#
# print('t1>>',t1) #结果:0.5146853923797607s 进程池的效率高
# print('t2>>',t2) #结果:12.092015027999878s
进程池的简单应用及与进程池的效率对比
map是异步执行的,并且自带close和join
一般约定俗成的是进程池中的进程数量为CPU的数量,工作中要看具体情况来考量。
一般约定俗成的是进程池中的进程数量为CPU的数量,工作中要看具体情况来考量。
进程池的同步调用
import os,time
from multiprocessing import Pool
def work(n):
print('%s run' %os.getpid())
time.sleep(1)
return n**2
if __name__ == '__main__':
p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
res_l=[]
for i in range(10):
res=p.apply(work,args=(i,)) # 同步调用,直到本次任务执行完毕拿到res,等待任务work执行的过程中可能有阻塞也可能没有阻塞
# 但不管该任务是否存在阻塞,同步调用都会在原地等着
res_l.append(res)
print(res_l)
进程池的同步调用
from multiprocessing import Pool
def work(n):
print('%s run' %os.getpid())
time.sleep(1)
return n**2
if __name__ == '__main__':
p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
res_l=[]
for i in range(10):
res=p.apply(work,args=(i,)) # 同步调用,直到本次任务执行完毕拿到res,等待任务work执行的过程中可能有阻塞也可能没有阻塞
# 但不管该任务是否存在阻塞,同步调用都会在原地等着
res_l.append(res)
print(res_l)
进程池的同步调用
进程池的异步调用
import os
import time
import random
from multiprocessing import Pool
def work(n):
print('%s run' %os.getpid())
time.sleep(random.random())
return n**2
if __name__ == '__main__':
p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
res_l=[]
for i in range(10):
res=p.apply_async(work,args=(i,)) # 异步运行,根据进程池中有的进程数,每次最多3个子进程在异步执行,并且可以执行不同的任务,传送任意的参数了。
# 返回结果之后,将结果放入列表,归还进程,之后再执行新的任务
# 需要注意的是,进程池中的三个进程不会同时开启或者同时结束
# 而是执行完一个就释放一个进程,这个进程就去接收新的任务。
res_l.append(res)
# 异步apply_async用法:如果使用异步提交的任务,主进程需要使用join,等待进程池内任务都处理完,然后可以用get收集结果
# 否则,主进程结束,进程池可能还没来得及执行,也就跟着一起结束了
p.close() #不是关闭进程池,而是结束进程池接收任务,确保没有新任务再提交过来。
p.join() #感知进程池中的任务已经执行结束,只有当没有新的任务添加进来的时候,才能感知到任务结束了,所以在join之前必须加上close方法
for res in res_l:
print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
进程池的异步调用
import time
import random
from multiprocessing import Pool
def work(n):
print('%s run' %os.getpid())
time.sleep(random.random())
return n**2
if __name__ == '__main__':
p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
res_l=[]
for i in range(10):
res=p.apply_async(work,args=(i,)) # 异步运行,根据进程池中有的进程数,每次最多3个子进程在异步执行,并且可以执行不同的任务,传送任意的参数了。
# 返回结果之后,将结果放入列表,归还进程,之后再执行新的任务
# 需要注意的是,进程池中的三个进程不会同时开启或者同时结束
# 而是执行完一个就释放一个进程,这个进程就去接收新的任务。
res_l.append(res)
# 异步apply_async用法:如果使用异步提交的任务,主进程需要使用join,等待进程池内任务都处理完,然后可以用get收集结果
# 否则,主进程结束,进程池可能还没来得及执行,也就跟着一起结束了
p.close() #不是关闭进程池,而是结束进程池接收任务,确保没有新任务再提交过来。
p.join() #感知进程池中的任务已经执行结束,只有当没有新的任务添加进来的时候,才能感知到任务结束了,所以在join之前必须加上close方法
for res in res_l:
print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
进程池的异步调用
详解:apply_async和apply
#一:使用进程池(异步调用,apply_async)
#coding: utf-8
from multiprocessing import Process,Pool
import time
def func(msg):
print( "msg:", msg)
time.sleep(1)
return msg
if __name__ == "__main__":
pool = Pool(processes = 3)
res_l=[]
for i in range(10):
msg = "hello %d" %(i)
res=pool.apply_async(func, (msg, )) #维持执行的进程总数为processes,当一个进程执行完毕后会添加新的进程进去
res_l.append(res)
# s = res.get() #如果直接用res这个结果对象调用get方法获取结果的话,这个程序就变成了同步,因为get方法直接就在这里等着你创建的进程的结果,第一个进程创建了,并且去执行了,那么get就会等着第一个进程的结果,没有结果就一直等着,那么主进程的for循环是无法继续的,所以你会发现变成了同步的效果
print("==============================>") #没有后面的join,或get,则程序整体结束,进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了
pool.close() #关闭进程池,防止进一步操作。如果所有操作持续挂起,它们将在工作进程终止前完成
pool.join() #调用join之前,先调用close函数,否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
print(res_l) #看到的是<multiprocessing.pool.ApplyResult object at 0x10357c4e0>对象组成的列表,而非最终的结果,但这一步是在join后执行的,证明结果已经计算完毕,剩下的事情就是调用每个对象下的get方法去获取结果
for i in res_l:
print(i.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
#二:使用进程池(同步调用,apply)
#coding: utf-8
from multiprocessing import Process,Pool
import time
def func(msg):
print( "msg:", msg)
time.sleep(0.1)
return msg
if __name__ == "__main__":
pool = Pool(processes = 3)
res_l=[]
for i in range(10):
msg = "hello %d" %(i)
res=pool.apply(func, (msg, )) #维持执行的进程总数为processes,当一个进程执行完毕后会添加新的进程进去
res_l.append(res) #同步执行,即执行完一个拿到结果,再去执行另外一个
print("==============================>")
pool.close()
pool.join() #调用join之前,先调用close函数,否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
print(res_l) #看到的就是最终的结果组成的列表
for i in res_l: #apply是同步的,所以直接得到结果,没有get()方法
print(i)
详解:apply_async和apply
#coding: utf-8
from multiprocessing import Process,Pool
import time
def func(msg):
print( "msg:", msg)
time.sleep(1)
return msg
if __name__ == "__main__":
pool = Pool(processes = 3)
res_l=[]
for i in range(10):
msg = "hello %d" %(i)
res=pool.apply_async(func, (msg, )) #维持执行的进程总数为processes,当一个进程执行完毕后会添加新的进程进去
res_l.append(res)
# s = res.get() #如果直接用res这个结果对象调用get方法获取结果的话,这个程序就变成了同步,因为get方法直接就在这里等着你创建的进程的结果,第一个进程创建了,并且去执行了,那么get就会等着第一个进程的结果,没有结果就一直等着,那么主进程的for循环是无法继续的,所以你会发现变成了同步的效果
print("==============================>") #没有后面的join,或get,则程序整体结束,进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了
pool.close() #关闭进程池,防止进一步操作。如果所有操作持续挂起,它们将在工作进程终止前完成
pool.join() #调用join之前,先调用close函数,否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
print(res_l) #看到的是<multiprocessing.pool.ApplyResult object at 0x10357c4e0>对象组成的列表,而非最终的结果,但这一步是在join后执行的,证明结果已经计算完毕,剩下的事情就是调用每个对象下的get方法去获取结果
for i in res_l:
print(i.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
#二:使用进程池(同步调用,apply)
#coding: utf-8
from multiprocessing import Process,Pool
import time
def func(msg):
print( "msg:", msg)
time.sleep(0.1)
return msg
if __name__ == "__main__":
pool = Pool(processes = 3)
res_l=[]
for i in range(10):
msg = "hello %d" %(i)
res=pool.apply(func, (msg, )) #维持执行的进程总数为processes,当一个进程执行完毕后会添加新的进程进去
res_l.append(res) #同步执行,即执行完一个拿到结果,再去执行另外一个
print("==============================>")
pool.close()
pool.join() #调用join之前,先调用close函数,否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
print(res_l) #看到的就是最终的结果组成的列表
for i in res_l: #apply是同步的,所以直接得到结果,没有get()方法
print(i)
详解:apply_async和apply
回调函数
需要回调函数的场景:进程池中任何一个任务一旦处理完了,就立即告知主进程:我好了额,你可以处理我的结果了。主进程则调用一个函数去处理该结果,该函数即回调函数,这是进程池特有的,普通进程没有这个机制,但是我们也可以通过进程通信来拿到返回值,进程池的这个回调也是进程通信的机制完成的。
我们可以把耗时间(阻塞)的任务放到进程池中,然后指定回调函数(主进程负责执行),这样主进程在执行回调函数时就省去了I/O的过程,直接拿到的是任务的结果
我们可以把耗时间(阻塞)的任务放到进程池中,然后指定回调函数(主进程负责执行),这样主进程在执行回调函数时就省去了I/O的过程,直接拿到的是任务的结果
import os
from multiprocessing import Pool
def func1(n):
print('func1>>',os.getpid())
print('func1')
return n*n
def func2(nn):
print('func2>>',os.getpid())
print('func2')
print(nn)
# import time
# time.sleep(0.5)
if __name__ == '__main__':
print('主进程:',os.getpid())
p = Pool(5)
#args里面的10给了func1,func1的返回值作为回调函数的参数给了callback对应的函数,不能直接给回调函数直接传参数,他只能是你任务函数func1的函数的返回值
# for i in range(10,20): #如果是多个进程来执行任务,那么当所有子进程将结果给了回调函数之后,回调函数又是在主进程上执行的,那么就会出现打印结果是同步的效果。我们上面func2里面注销的时间模块打开看看
# p.apply_async(func1,args=(i,),callback=func2)
p.apply_async(func1,args=(10,),callback=func2)
p.close()
p.join()
#结果
# 主进程: 11852 #发现回调函数是在主进程中完成的,其实如果是在子进程中完成的,那我们直接将代码写在子进程的任务函数func1里面就行了,对不对,这也是为什么称为回调函数的原因。
# func1>> 17332
# func1
# func2>> 11852
# func2
# 100
回调函数的简单使用
from multiprocessing import Pool
def func1(n):
print('func1>>',os.getpid())
print('func1')
return n*n
def func2(nn):
print('func2>>',os.getpid())
print('func2')
print(nn)
# import time
# time.sleep(0.5)
if __name__ == '__main__':
print('主进程:',os.getpid())
p = Pool(5)
#args里面的10给了func1,func1的返回值作为回调函数的参数给了callback对应的函数,不能直接给回调函数直接传参数,他只能是你任务函数func1的函数的返回值
# for i in range(10,20): #如果是多个进程来执行任务,那么当所有子进程将结果给了回调函数之后,回调函数又是在主进程上执行的,那么就会出现打印结果是同步的效果。我们上面func2里面注销的时间模块打开看看
# p.apply_async(func1,args=(i,),callback=func2)
p.apply_async(func1,args=(10,),callback=func2)
p.close()
p.join()
#结果
# 主进程: 11852 #发现回调函数是在主进程中完成的,其实如果是在子进程中完成的,那我们直接将代码写在子进程的任务函数func1里面就行了,对不对,这也是为什么称为回调函数的原因。
# func1>> 17332
# func1
# func2>> 11852
# func2
# 100
回调函数的简单使用
线程Threads
背景知识
之前我们已经了解了操作系统中进程的概念,程序并不能单独运行,只有将程序装载到内存中,系统为它分配资源才能运行,而这种执行的程序就称之为进程。程序和进程的区别就在于:程序是指令的集合,它是进程运行的静态描述文本;进程是程序的一次执行活动,属于动态概念。在多道编程中,我们允许多个程序同时加载到内存中,在操作系统的调度下,可以实现并发地执行。这是这样的设计,大大提高了CPU的利用率。进程的出现让每个用户感觉到自己独享CPU,因此,进程就是为了在CPU上实现多道编程而提出的。
进程的不足
进程有很多优点,它提供了多道编程,让我们感觉我们每个人都拥有自己的CPU和其他资源,可以提高计算机的利用率。很多人就不理解了,既然进程这么优秀,为什么还要线程呢?其实,仔细观察就会发现进程还是有很多缺陷的,主要体现在两点上:
进程只能在一个时间干一件事,如果想同时干两件事或多件事,进程就无能为力了。
进程在执行的过程中如果阻塞,例如等待输入,整个进程就会挂起,即使进程中有些工作不依赖于输入的数据,也将无法执行。
进程只能在一个时间干一件事,如果想同时干两件事或多件事,进程就无能为力了。
进程在执行的过程中如果阻塞,例如等待输入,整个进程就会挂起,即使进程中有些工作不依赖于输入的数据,也将无法执行。
举例说明: 如果这两个缺点理解比较困难的话,举个现实的例子也许你就清楚了:如果把我们上课的过程看成一个进程的话,那么我们要做的是耳朵听老师讲课,手上还要记笔记,脑子还要思考问题,这样才能高效的完成听课的任务。而如果只提供进程这个机制的话,上面这三件事将不能同时执行,同一时间只能做一件事,听的时候就不能记笔记,也不能用脑子思考,这是其一;如果老师在黑板上写演算过程,我们开始记笔记,而老师突然有一步推不下去了,阻塞住了,他在那边思考着,而我们呢,也不能干其他事,即使你想趁此时思考一下刚才没听懂的一个问题都不行,这是其二。
现在你应该明白了进程的缺陷了,而解决的办法很简单,我们完全可以让听、写、思三个独立的过程,并行起来,这样很明显可以提高听课的效率。而实际的操作系统中,也同样引入了这种类似的机制——线程。
3.线程的出现
现在你应该明白了进程的缺陷了,而解决的办法很简单,我们完全可以让听、写、思三个独立的过程,并行起来,这样很明显可以提高听课的效率。而实际的操作系统中,也同样引入了这种类似的机制——线程。
3.线程的出现
什么线程
指的是一条流水线的工作过程,关键的一句话:一个进程内最少自带一个线程,其实进程根本不能执行,进程不是执行单位,是资源的单位,分配资源的单位
注意: 在80年代,出现了能独立运行的基本单位——线程(Threads)。
在80年代,出现了能独立运行的基本单位——线程(Threads)。
每一个进程中至少有一个线程。
注意:进程是资源分配的最小单位,线程是CPU调度的最小单位.
线程的特征
1)轻型实体
线程中的实体基本上不拥有系统资源,只是有一些必不可少的、能保证独立运行的资源。
线程的实体包括程序、数据和TCB。线程是动态概念,它的动态特性由线程控制块TCB(Thread Control Block)描述。
线程的实体包括程序、数据和TCB。线程是动态概念,它的动态特性由线程控制块TCB(Thread Control Block)描述。
2)独立调度和分派的基本单位。
在多线程OS中,线程是能独立运行的基本单位,因而也是独立调度和分派的基本单位。由于线程很“轻”,
故线程的切换非常迅速且开销小(在同一进程中的)。
故线程的切换非常迅速且开销小(在同一进程中的)。
3)共享进程资源
线程在同一进程中的各个线程,都可以共享该进程所拥有的资源,这首先表现在:所有线程都具有相同的进程id,这意味着,线程可以访问该进程的每一个内存资源;此外,还可以访问进程所拥有的已打开文件、定时器、信号量机构等。由于同一个进程内的线程共享内存和文件,所以线程之间互相通信不必调用内核。
4)可并发执行。
在一个进程中的多个线程之间,可以并发执行,甚至允许在一个进程中所有线程都能并发执行;
同样,不同进程中的线程也能并发执行,充分利用和发挥了处理机与外围设备并行工作的能力。
同样,不同进程中的线程也能并发执行,充分利用和发挥了处理机与外围设备并行工作的能力。
内存中的线程
多个线程共享同一个进程的地址空间中的资源,是对一台计算机上多个进程的模拟,有时也称线程为轻量级的进程。
而对一台计算机上多个进程,则共享物理内存、磁盘、打印机等其他物理资源。多线程的运行也多进程的运行类似,是cpu在多个线程之间的快速切换。
不同的进程之间是充满敌意的,彼此是抢占、竞争cpu的关系,如果迅雷会和QQ抢资源。而同一个进程是由一个程序员的程序创建,所以同一进程内的线程是合作关系,一个线程可以访问另外一个线程的内存地址,大家都是共享的,一个线程干死了另外一个线程的内存,那纯属程序员脑子有问题。
类似于进程,每个线程也有自己的堆栈,不同于进程,线程库无法利用时钟中断强制线程让出CPU,可以调用thread_yield运行线程自动放弃cpu,让另外一个线程运行。
线程通常是有益的,但是带来了不小程序设计难度,线程的问题是:
1. 父进程有多个线程,那么开启的子线程是否需要同样多的线程
2. 在同一个进程中,如果一个线程关闭了文件,而另外一个线程正准备往该文件内写内容呢?
因此,在多线程的代码中,需要更多的心思来设计程序的逻辑、保护程序的数据。
而对一台计算机上多个进程,则共享物理内存、磁盘、打印机等其他物理资源。多线程的运行也多进程的运行类似,是cpu在多个线程之间的快速切换。
不同的进程之间是充满敌意的,彼此是抢占、竞争cpu的关系,如果迅雷会和QQ抢资源。而同一个进程是由一个程序员的程序创建,所以同一进程内的线程是合作关系,一个线程可以访问另外一个线程的内存地址,大家都是共享的,一个线程干死了另外一个线程的内存,那纯属程序员脑子有问题。
类似于进程,每个线程也有自己的堆栈,不同于进程,线程库无法利用时钟中断强制线程让出CPU,可以调用thread_yield运行线程自动放弃cpu,让另外一个线程运行。
线程通常是有益的,但是带来了不小程序设计难度,线程的问题是:
1. 父进程有多个线程,那么开启的子线程是否需要同样多的线程
2. 在同一个进程中,如果一个线程关闭了文件,而另外一个线程正准备往该文件内写内容呢?
因此,在多线程的代码中,需要更多的心思来设计程序的逻辑、保护程序的数据。
用户级线程和内核级线程(了解)
.用户级线程
内核的切换由用户态程序自己控制内核切换,不需要内核干涉,少了进出内核态的消耗,但不能很好的利用多核Cpu。
内核的切换由用户态程序自己控制内核切换,不需要内核干涉,少了进出内核态的消耗,但不能很好的利用多核Cpu。
内核级线程
内核级线程:切换由内核控制,当线程进行切换的时候,由用户态转化为内核态。切换完毕要从内核态返回用户态;可以很好的利用smp,即利用多核cpu。windows线程就是这样的。
内核级线程:切换由内核控制,当线程进行切换的时候,由用户态转化为内核态。切换完毕要从内核态返回用户态;可以很好的利用smp,即利用多核cpu。windows线程就是这样的。
混合实现
用户级与内核级的多路复用,内核同一调度内核线程,每个内核线程对应n个用户线程,用户和内核都能感知到的线程,用户创建一个线程,那么操作系统内核也跟着创建一个线程来专门执行你用户的这个线程。
用户级与内核级的多路复用,内核同一调度内核线程,每个内核线程对应n个用户线程,用户和内核都能感知到的线程,用户创建一个线程,那么操作系统内核也跟着创建一个线程来专门执行你用户的这个线程。
Threading模块
进程与线程开启效率比较
from threading import Thread
from multiprocessing import Process
import os
import time
def work():
print('hello')
if __name__ == '__main__':
s1 = time.time()
#在主进程下开启线程
t=Thread(target=work)
t.start()
t.join()
t1 = time.time() - s1
print('进程的执行时间:',t1)
print('主线程/主进程')
'''
打印结果:
hello
进程的执行时间: 0.0
主线程/主进程
'''
s2 = time.time()
#在主进程下开启子进程
t=Process(target=work)
t.start()
t.join()
t2 = time.time() - s2
print('线程的执行时间:', t2)
print('主线程/主进程')
'''
打印结果:
hello
线程的执行时间: 0.5216977596282959
主线程/主进程
'''
进程与线程开启效率比较
from multiprocessing import Process
import os
import time
def work():
print('hello')
if __name__ == '__main__':
s1 = time.time()
#在主进程下开启线程
t=Thread(target=work)
t.start()
t.join()
t1 = time.time() - s1
print('进程的执行时间:',t1)
print('主线程/主进程')
'''
打印结果:
hello
进程的执行时间: 0.0
主线程/主进程
'''
s2 = time.time()
#在主进程下开启子进程
t=Process(target=work)
t.start()
t.join()
t2 = time.time() - s2
print('线程的执行时间:', t2)
print('主线程/主进程')
'''
打印结果:
hello
线程的执行时间: 0.5216977596282959
主线程/主进程
'''
进程与线程开启效率比较
内存数据共享
from threading import Thread
from multiprocessing import Process
import os
def work():
global n #修改全局变量的值
n=0
if __name__ == '__main__':
# n=100
# p=Process(target=work)
# p.start()
# p.join()
# print('主',n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100
n=1
t=Thread(target=work)
t.start()
t.join() #必须加join,因为主线程和子线程不一定谁快,一般都是主线程快一些,所有我们要等子线程执行完毕才能看出效果
print('主',n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据
# 通过一个global就实现了全局变量的使用,不需要进程的IPC通信方法
内存数据共享
from multiprocessing import Process
import os
def work():
global n #修改全局变量的值
n=0
if __name__ == '__main__':
# n=100
# p=Process(target=work)
# p.start()
# p.join()
# print('主',n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100
n=1
t=Thread(target=work)
t.start()
t.join() #必须加join,因为主线程和子线程不一定谁快,一般都是主线程快一些,所有我们要等子线程执行完毕才能看出效果
print('主',n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据
# 通过一个global就实现了全局变量的使用,不需要进程的IPC通信方法
内存数据共享
线程的创建
直接上代码(和进程创建类似)
方式一
方式一
方式二
from threading import Thread
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
if __name__ == '__main__':
t=Thread(target=sayhi,args=('太白',))
t.start()
print('主线程')
方式1
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
if __name__ == '__main__':
t=Thread(target=sayhi,args=('太白',))
t.start()
print('主线程')
方式1
import time
from threading import Thread
class Sayhi(Thread):
def __init__(self,name):
super().__init__()
self.name=name
def run(self):
time.sleep(2)
print('%s say hello' % self.name)
if __name__ == '__main__':
t = Sayhi('太白')
t.start()
print('主线程')
方式2
from threading import Thread
class Sayhi(Thread):
def __init__(self,name):
super().__init__()
self.name=name
def run(self):
time.sleep(2)
print('%s say hello' % self.name)
if __name__ == '__main__':
t = Sayhi('太白')
t.start()
print('主线程')
方式2
Threads的其他方法
Thread实例对象的方法
# isAlive(): 返回线程是否活动的。
# getName(): 返回线程名。
# setName(): 设置线程名。
threading模块提供的一些方法:
# threading.currentThread(): 返回当前的线程变量。
# threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前,不包括启动前和终止后的线程。
# threading.activeCount(): 返回正在运行的线程数量,与len(threading.enumerate())有相同的结果
复制代码
# isAlive(): 返回线程是否活动的。
# getName(): 返回线程名。
# setName(): 设置线程名。
threading模块提供的一些方法:
# threading.currentThread(): 返回当前的线程变量。
# threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前,不包括启动前和终止后的线程。
# threading.activeCount(): 返回正在运行的线程数量,与len(threading.enumerate())有相同的结果
复制代码
join方法
from threading import Thread
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
if __name__ == '__main__':
t=Thread(target=sayhi,args=('太白',))
t2=Thread(target=sayhi,args=('alex',))
t.start()
t2.start()
t.join() #因为这个线程用了join方法,主线程等待子线程的运行结束
print('主线程')
print(t.is_alive()) #所以t这个线程肯定是执行结束了,结果为False
print(t2.is_alive()) #有可能是True,有可能是False,看子线程和主线程谁执行的快
'''
egon say hello
主线程
False
'''
join方法
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
if __name__ == '__main__':
t=Thread(target=sayhi,args=('太白',))
t2=Thread(target=sayhi,args=('alex',))
t.start()
t2.start()
t.join() #因为这个线程用了join方法,主线程等待子线程的运行结束
print('主线程')
print(t.is_alive()) #所以t这个线程肯定是执行结束了,结果为False
print(t2.is_alive()) #有可能是True,有可能是False,看子线程和主线程谁执行的快
'''
egon say hello
主线程
False
'''
join方法
守护线程 #1.对主进程来说,运行完毕指的是主进程代码运行完毕
#1.对主进程来说,运行完毕指的是主进程代码运行完毕
#2.对主线程来说,运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕,主线程才算运行完毕
示例1:
from threading import Thread
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
if __name__ == '__main__':
t=Thread(target=sayhi,args=('taibai',))
t.setDaemon(True) #必须在t.start()之前设置
t.start()
print('主线程')
print(t.is_alive())
'''
主线程
True
'''
守护线程示例1
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
if __name__ == '__main__':
t=Thread(target=sayhi,args=('taibai',))
t.setDaemon(True) #必须在t.start()之前设置
t.start()
print('主线程')
print(t.is_alive())
'''
主线程
True
'''
守护线程示例1
示例二
from threading import Thread
from multiprocessing import Process
import time
def func1():
while True:
print(666)
time.sleep(0.5)
def func2():
print('hello')
time.sleep(3)
if __name__ == '__main__':
# t = Thread(target=func1,)
# t.daemon = True #主线程结束,守护线程随之结束
# # t.setDaemon(True) #两种方式,和上面设置守护线程是一样的
# t.start()
# t2 = Thread(target=func2,) #这个子线程要执行3秒,主线程的代码虽然执行完了,但是一直等着子线程的任务执行完毕,主线程才算完毕,因为通过结果你会发现我主线程虽然代码执行完毕了,\
# 但是主线程的的守护线程t1还在执行,说明什么,说明我的主线程还没有完毕,只不过是代码执行完了,一直等着子线程t2执行完毕,我主线程的守护线程才停止,说明子线程执行完毕之后,我的主线程才执行完毕
# t2.start()
# print('主线程代码执行完啦!')
p = Process(target=func1,)
p.daemon = True
p.start()
p2 = Process(target=func2,)
p2.start()
time.sleep(1) #让主进程等1秒,为了能看到func1的打印效果
print('主进程代码执行完啦!') #通过结果你会发现,如果主进程的代码运行完毕了,那么主进程就结束了,因为主进程的守护进程p随着主进程的代码结束而结束了,守护进程被回收了,这和线程是不一样的,主线程的代码完了并不代表主线程运行完毕了,需要等着所有其他的非守护的子线程执行完毕才算完毕
守护线程示例2
from multiprocessing import Process
import time
def func1():
while True:
print(666)
time.sleep(0.5)
def func2():
print('hello')
time.sleep(3)
if __name__ == '__main__':
# t = Thread(target=func1,)
# t.daemon = True #主线程结束,守护线程随之结束
# # t.setDaemon(True) #两种方式,和上面设置守护线程是一样的
# t.start()
# t2 = Thread(target=func2,) #这个子线程要执行3秒,主线程的代码虽然执行完了,但是一直等着子线程的任务执行完毕,主线程才算完毕,因为通过结果你会发现我主线程虽然代码执行完毕了,\
# 但是主线程的的守护线程t1还在执行,说明什么,说明我的主线程还没有完毕,只不过是代码执行完了,一直等着子线程t2执行完毕,我主线程的守护线程才停止,说明子线程执行完毕之后,我的主线程才执行完毕
# t2.start()
# print('主线程代码执行完啦!')
p = Process(target=func1,)
p.daemon = True
p.start()
p2 = Process(target=func2,)
p2.start()
time.sleep(1) #让主进程等1秒,为了能看到func1的打印效果
print('主进程代码执行完啦!') #通过结果你会发现,如果主进程的代码运行完毕了,那么主进程就结束了,因为主进程的守护进程p随着主进程的代码结束而结束了,守护进程被回收了,这和线程是不一样的,主线程的代码完了并不代表主线程运行完毕了,需要等着所有其他的非守护的子线程执行完毕才算完毕
守护线程示例2
锁
.GIL锁(Global Interpreter Lock)
GIL并不是Python的特性,它是在实现Python解析器(CPython)时所引入的一个概念。
Python完全可以不依赖于GIL
GIL本质就是一把互斥锁,既然是互斥锁,所有互斥锁的本质都一样,都是将并发运行变成串行,
以此来控制同一时间内共享数据只能被一个任务所修改,进而保证数据安全。
可以肯定的一点是:保护不同的数据的安全,就应该加不同的锁。
要想了解GIL,首先确定一点:每次执行python程序,都会产生一个独立的进程。
例如python test.py,python aaa.py,python bbb.py会产生3个不同的python进程
以此来控制同一时间内共享数据只能被一个任务所修改,进而保证数据安全。
可以肯定的一点是:保护不同的数据的安全,就应该加不同的锁。
要想了解GIL,首先确定一点:每次执行python程序,都会产生一个独立的进程。
例如python test.py,python aaa.py,python bbb.py会产生3个不同的python进程
GIL锁与互斥锁综合分析(重点)
分析:
#1.100个线程去抢GIL锁,即抢执行权限
#2. 肯定有一个线程先抢到GIL(暂且称为线程1),然后开始执行,一旦执行就会拿到lock.acquire()
#3. 极有可能线程1还未运行完毕,就有另外一个线程2抢到GIL,然后开始运行,但线程2发现互斥锁lock还未被线程1释放,于是阻塞,被迫交出执行权限,即释放GIL
#4.直到线程1重新抢到GIL,开始从上次暂停的位置继续执行,直到正常释放互斥锁lock,然后其他的线程再重复2 3 4的过程
#1.100个线程去抢GIL锁,即抢执行权限
#2. 肯定有一个线程先抢到GIL(暂且称为线程1),然后开始执行,一旦执行就会拿到lock.acquire()
#3. 极有可能线程1还未运行完毕,就有另外一个线程2抢到GIL,然后开始运行,但线程2发现互斥锁lock还未被线程1释放,于是阻塞,被迫交出执行权限,即释放GIL
#4.直到线程1重新抢到GIL,开始从上次暂停的位置继续执行,直到正常释放互斥锁lock,然后其他的线程再重复2 3 4的过程
互斥锁与join的区别(重点)
#不加锁:并发执行,速度快,数据不安全
from threading import current_thread,Thread,Lock
import os,time
def task():
global n
print('%s is running' %current_thread().getName())
temp=n
time.sleep(0.5)
n=temp-1
if __name__ == '__main__':
n=100
lock=Lock()
threads=[]
start_time=time.time()
for i in range(100):
t=Thread(target=task)
threads.append(t)
t.start()
for t in threads:
t.join()
stop_time=time.time()
print('主:%s n:%s' %(stop_time-start_time,n))
'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:0.5216062068939209 n:99
'''
#不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全
from threading import current_thread,Thread,Lock
import os,time
def task():
#未加锁的代码并发运行
time.sleep(3)
print('%s start to run' %current_thread().getName())
global n
#加锁的代码串行运行
lock.acquire()
temp=n
time.sleep(0.5)
n=temp-1
lock.release()
if __name__ == '__main__':
n=100
lock=Lock()
threads=[]
start_time=time.time()
for i in range(100):
t=Thread(target=task)
threads.append(t)
t.start()
for t in threads:
t.join()
stop_time=time.time()
print('主:%s n:%s' %(stop_time-start_time,n))
'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:53.294203758239746 n:0
'''
#有的同学可能有疑问:既然加锁会让运行变成串行,那么我在start之后立即使用join,就不用加锁了啊,也是串行的效果啊
#没错:在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是
#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的
#单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.
from threading import current_thread,Thread,Lock
import os,time
def task():
time.sleep(3)
print('%s start to run' %current_thread().getName())
global n
temp=n
time.sleep(0.5)
n=temp-1
if __name__ == '__main__':
n=100
lock=Lock()
start_time=time.time()
for i in range(100):
t=Thread(target=task)
t.start()
t.join()
stop_time=time.time()
print('主:%s n:%s' %(stop_time-start_time,n))
'''
Thread-1 start to run
Thread-2 start to run
......
Thread-100 start to run
主:350.6937336921692 n:0 #耗时是多么的恐怖
'''
互斥锁与join的区别(重点)
from threading import current_thread,Thread,Lock
import os,time
def task():
global n
print('%s is running' %current_thread().getName())
temp=n
time.sleep(0.5)
n=temp-1
if __name__ == '__main__':
n=100
lock=Lock()
threads=[]
start_time=time.time()
for i in range(100):
t=Thread(target=task)
threads.append(t)
t.start()
for t in threads:
t.join()
stop_time=time.time()
print('主:%s n:%s' %(stop_time-start_time,n))
'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:0.5216062068939209 n:99
'''
#不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全
from threading import current_thread,Thread,Lock
import os,time
def task():
#未加锁的代码并发运行
time.sleep(3)
print('%s start to run' %current_thread().getName())
global n
#加锁的代码串行运行
lock.acquire()
temp=n
time.sleep(0.5)
n=temp-1
lock.release()
if __name__ == '__main__':
n=100
lock=Lock()
threads=[]
start_time=time.time()
for i in range(100):
t=Thread(target=task)
threads.append(t)
t.start()
for t in threads:
t.join()
stop_time=time.time()
print('主:%s n:%s' %(stop_time-start_time,n))
'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:53.294203758239746 n:0
'''
#有的同学可能有疑问:既然加锁会让运行变成串行,那么我在start之后立即使用join,就不用加锁了啊,也是串行的效果啊
#没错:在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是
#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的
#单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.
from threading import current_thread,Thread,Lock
import os,time
def task():
time.sleep(3)
print('%s start to run' %current_thread().getName())
global n
temp=n
time.sleep(0.5)
n=temp-1
if __name__ == '__main__':
n=100
lock=Lock()
start_time=time.time()
for i in range(100):
t=Thread(target=task)
t.start()
t.join()
stop_time=time.time()
print('主:%s n:%s' %(stop_time-start_time,n))
'''
Thread-1 start to run
Thread-2 start to run
......
Thread-100 start to run
主:350.6937336921692 n:0 #耗时是多么的恐怖
'''
互斥锁与join的区别(重点)
.同步锁
三个需要注意的点:
#1.线程抢的是GIL锁,GIL锁相当于执行权限,拿到执行权限后才能拿到互斥锁Lock,其他线程也可以抢到GIL,但如果发现Lock仍然没有被释放则阻塞,即便是拿到执行权限GIL也要立刻交出来
#2.join是等待所有,即整体串行,而锁只是锁住修改共享数据的部分,即部分串行,要想保证数据安全的根本原理在于让并发变成串行,join与互斥锁都可以实现,毫无疑问,互斥锁的部分串行效率要更高
#3. 一定要看本小节最后的GIL与互斥锁的经典分析
#1.线程抢的是GIL锁,GIL锁相当于执行权限,拿到执行权限后才能拿到互斥锁Lock,其他线程也可以抢到GIL,但如果发现Lock仍然没有被释放则阻塞,即便是拿到执行权限GIL也要立刻交出来
#2.join是等待所有,即整体串行,而锁只是锁住修改共享数据的部分,即部分串行,要想保证数据安全的根本原理在于让并发变成串行,join与互斥锁都可以实现,毫无疑问,互斥锁的部分串行效率要更高
#3. 一定要看本小节最后的GIL与互斥锁的经典分析
.死锁与递归锁
进程也有死锁与递归锁,在进程那里忘记说了,放到这里一切说了额,进程的死锁和线程的是一样的,而且一般情况下进程之间是数据不共享的,不需要加锁,由于线程是对全局的数据共享的,所以对于全局的数据进行操作的时候,要加锁。
所谓死锁: 是指两个或两个以上的进程或线程在执行过程中,因争夺资源而造成的一种互相等待的现象,若无外力作用,它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁,这些永远在互相等待的进程称为死锁进程,如下就是死锁
所谓死锁: 是指两个或两个以上的进程或线程在执行过程中,因争夺资源而造成的一种互相等待的现象,若无外力作用,它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁,这些永远在互相等待的进程称为死锁进程,如下就是死锁
from threading import Lock as Lock
import time
mutexA=Lock()
mutexA.acquire()
mutexA.acquire()
print(123)
mutexA.release()
mutexA.release()
import time
mutexA=Lock()
mutexA.acquire()
mutexA.acquire()
print(123)
mutexA.release()
mutexA.release()
递归锁解决死锁问题
import time
from threading import Thread,RLock
fork_lock = noodle_lock = RLock()
def eat1(name):
noodle_lock.acquire()
print('%s 抢到了面条'%name)
fork_lock.acquire()
print('%s 抢到了叉子'%name)
print('%s 吃面'%name)
fork_lock.release()
noodle_lock.release()
def eat2(name):
fork_lock.acquire()
print('%s 抢到了叉子' % name)
time.sleep(1)
noodle_lock.acquire()
print('%s 抢到了面条' % name)
print('%s 吃面' % name)
noodle_lock.release()
fork_lock.release()
for name in ['taibai','wulaoban']:
t1 = Thread(target=eat1,args=(name,))
t1.start()
for name in ['alex','peiqi']:
t2 = Thread(target=eat2,args=(name,))
t2.start()
递归锁解决死锁问题
from threading import Thread,RLock
fork_lock = noodle_lock = RLock()
def eat1(name):
noodle_lock.acquire()
print('%s 抢到了面条'%name)
fork_lock.acquire()
print('%s 抢到了叉子'%name)
print('%s 吃面'%name)
fork_lock.release()
noodle_lock.release()
def eat2(name):
fork_lock.acquire()
print('%s 抢到了叉子' % name)
time.sleep(1)
noodle_lock.acquire()
print('%s 抢到了面条' % name)
print('%s 吃面' % name)
noodle_lock.release()
fork_lock.release()
for name in ['taibai','wulaoban']:
t1 = Thread(target=eat1,args=(name,))
t1.start()
for name in ['alex','peiqi']:
t2 = Thread(target=eat2,args=(name,))
t2.start()
递归锁解决死锁问题
递归锁大致描述: 当我们的程序中需要两把锁的时候,你就要注意,别出现死锁,最好就去用递归锁。
信号量
同进程的一样
Semaphore管理一个内置的计数器,
每当调用acquire()时内置计数器-1;
调用release() 时内置计数器+1;
计数器不能小于0;当计数器为0时,acquire()将阻塞线程直到其他线程调用release()。
实例:(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5):
Semaphore管理一个内置的计数器,
每当调用acquire()时内置计数器-1;
调用release() 时内置计数器+1;
计数器不能小于0;当计数器为0时,acquire()将阻塞线程直到其他线程调用release()。
实例:(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5):
from threading import Thread,Semaphore
import threading
import time
# def func():
# if sm.acquire():
# print (threading.currentThread().getName() + ' get semaphore')
# time.sleep(2)
# sm.release()
def func():
sm.acquire()
print('%s get sm' %threading.current_thread().getName())
time.sleep(3)
sm.release()
if __name__ == '__main__':
sm=Semaphore(5)
for i in range(23):
t=Thread(target=func)
t.start()
示例代码
import threading
import time
# def func():
# if sm.acquire():
# print (threading.currentThread().getName() + ' get semaphore')
# time.sleep(2)
# sm.release()
def func():
sm.acquire()
print('%s get sm' %threading.current_thread().getName())
time.sleep(3)
sm.release()
if __name__ == '__main__':
sm=Semaphore(5)
for i in range(23):
t=Thread(target=func)
t.start()
示例代码
事件
事件的基本方法:
event.isSet():返回event的状态值;
event.wait():如果 event.isSet()==False将阻塞线程;
event.set(): 设置event的状态值为True,所有阻塞池的线程激活进入就绪状态, 等待操作系统调度;
event.clear():恢复event的状态值为False。
event.isSet():返回event的状态值;
event.wait():如果 event.isSet()==False将阻塞线程;
event.set(): 设置event的状态值为True,所有阻塞池的线程激活进入就绪状态, 等待操作系统调度;
event.clear():恢复event的状态值为False。
条件Condition(了解)
使得线程等待,只有满足某条件时,才释放n个线程,看一下大概怎么用就可以啦~~
import time
from threading import Thread,RLock,Condition,current_thread
def func1(c):
c.acquire(False) #固定格式
# print(1111)
c.wait() #等待通知,
time.sleep(3) #通知完成后大家是串行执行的,这也看出了锁的机制了
print('%s执行了'%(current_thread().getName()))
c.release()
if __name__ == '__main__':
c = Condition()
for i in range(5):
t = Thread(target=func1,args=(c,))
t.start()
while True:
num = int(input('请输入你要通知的线程个数:'))
c.acquire() #固定格式
c.notify(num) #通知num个线程别等待了,去执行吧
c.release()
#结果分析:
# 请输入你要通知的线程个数:3
# 请输入你要通知的线程个数:Thread-1执行了 #有时候你会发现的你结果打印在了你要输入内容的地方,这是打印的问题,没关系,不影响
# Thread-3执行了
# Thread-2执行了
示例代码
from threading import Thread,RLock,Condition,current_thread
def func1(c):
c.acquire(False) #固定格式
# print(1111)
c.wait() #等待通知,
time.sleep(3) #通知完成后大家是串行执行的,这也看出了锁的机制了
print('%s执行了'%(current_thread().getName()))
c.release()
if __name__ == '__main__':
c = Condition()
for i in range(5):
t = Thread(target=func1,args=(c,))
t.start()
while True:
num = int(input('请输入你要通知的线程个数:'))
c.acquire() #固定格式
c.notify(num) #通知num个线程别等待了,去执行吧
c.release()
#结果分析:
# 请输入你要通知的线程个数:3
# 请输入你要通知的线程个数:Thread-1执行了 #有时候你会发现的你结果打印在了你要输入内容的地方,这是打印的问题,没关系,不影响
# Thread-3执行了
# Thread-2执行了
示例代码
定时器(了解)
定时器,指定n秒后执行某个操作,这个做定时任务的时候可能会用到。
import time
from threading import Timer,current_thread #这里就不需要再引入Timer
import threading
def hello():
print(current_thread().getName())
print("hello, world")
# time.sleep(3) #如果你的子线程的程序执行时间比较长,那么这个定时任务也会乱,当然了,主要还是看业务需求
t = Timer(10, hello) #创建一个子线程去执行后面的函数
t.start() # after 1 seconds, "hello, world" will be printed
# for i in range(5):
# t = Timer(2, hello)
# t.start()
# time.sleep(3) #这个是创建一个t用的时间是2秒,创建出来第二个的时候,第一个已经过了两秒了,所以你的5个t的执行结果基本上就是2秒中,这个延迟操作。
print(threading.active_count())
print('主进程',current_thread().getName())
代码示例
from threading import Timer,current_thread #这里就不需要再引入Timer
import threading
def hello():
print(current_thread().getName())
print("hello, world")
# time.sleep(3) #如果你的子线程的程序执行时间比较长,那么这个定时任务也会乱,当然了,主要还是看业务需求
t = Timer(10, hello) #创建一个子线程去执行后面的函数
t.start() # after 1 seconds, "hello, world" will be printed
# for i in range(5):
# t = Timer(2, hello)
# t.start()
# time.sleep(3) #这个是创建一个t用的时间是2秒,创建出来第二个的时候,第一个已经过了两秒了,所以你的5个t的执行结果基本上就是2秒中,这个延迟操作。
print(threading.active_count())
print('主进程',current_thread().getName())
代码示例
线程队列
线程之间的通信我们列表行不行呢,当然行,那么队列和列表有什么区别呢?
queue队列 :使用import queue,用法与进程Queue一样
queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.
class queue.Queue(maxsize=0) #先进先出
queue队列 :使用import queue,用法与进程Queue一样
queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.
class queue.Queue(maxsize=0) #先进先出
先进先出队列代码
class queue.Queue(maxsize=0) #先进先出
class queue.Queue(maxsize=0) #先进先出
import queue #不需要通过threading模块里面导入,直接import queue就可以了,这是python自带的
#用法基本和我们进程multiprocess中的queue是一样的
q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')
# q.put_nowait() #没有数据就报错,可以通过try来搞
print(q.get())
print(q.get())
print(q.get())
# q.get_nowait() #没有数据就报错,可以通过try来搞
'''
结果(先进先出):
first
second
third
'''
先进先出示例代码
#用法基本和我们进程multiprocess中的queue是一样的
q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')
# q.put_nowait() #没有数据就报错,可以通过try来搞
print(q.get())
print(q.get())
print(q.get())
# q.get_nowait() #没有数据就报错,可以通过try来搞
'''
结果(先进先出):
first
second
third
'''
先进先出示例代码
先进后出示例代码
class queue.LifoQueue(maxsize=0) #last in fisrt out
class queue.LifoQueue(maxsize=0) #last in fisrt out
import queue
q=queue.LifoQueue() #队列,类似于栈,栈我们提过吗,是不是先进后出的顺序啊
q.put('first')
q.put('second')
q.put('third')
# q.put_nowait()
print(q.get())
print(q.get())
print(q.get())
# q.get_nowait()
'''
结果(后进先出):
third
second
first
'''
先进后出示例代码
q=queue.LifoQueue() #队列,类似于栈,栈我们提过吗,是不是先进后出的顺序啊
q.put('first')
q.put('second')
q.put('third')
# q.put_nowait()
print(q.get())
print(q.get())
print(q.get())
# q.get_nowait()
'''
结果(后进先出):
third
second
first
'''
先进后出示例代码
优先级队列示例代码
class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列
class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列
import queue
q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
q.put((-10,'a'))
q.put((-5,'a')) #负数也可以
# q.put((20,'ws')) #如果两个值的优先级一样,那么按照后面的值的acsii码顺序来排序,如果字符串第一个数元素相同,比较第二个元素的acsii码顺序
# q.put((20,'wd'))
# q.put((20,{'a':11})) #TypeError: unorderable types: dict() < dict() 不能是字典
# q.put((20,('w',1))) #优先级相同的两个数据,他们后面的值必须是相同的数据类型才能比较,可以是元祖,也是通过元素的ascii码顺序来排序
q.put((20,'b'))
q.put((20,'a'))
q.put((0,'b'))
q.put((30,'c'))
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
'''
结果(数字越小优先级越高,优先级高的优先出队):
'''
优先级队列示例代码
q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
q.put((-10,'a'))
q.put((-5,'a')) #负数也可以
# q.put((20,'ws')) #如果两个值的优先级一样,那么按照后面的值的acsii码顺序来排序,如果字符串第一个数元素相同,比较第二个元素的acsii码顺序
# q.put((20,'wd'))
# q.put((20,{'a':11})) #TypeError: unorderable types: dict() < dict() 不能是字典
# q.put((20,('w',1))) #优先级相同的两个数据,他们后面的值必须是相同的数据类型才能比较,可以是元祖,也是通过元素的ascii码顺序来排序
q.put((20,'b'))
q.put((20,'a'))
q.put((0,'b'))
q.put((30,'c'))
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
'''
结果(数字越小优先级越高,优先级高的优先出队):
'''
优先级队列示例代码
这三种队列都是线程安全的,不会出现多个线程抢占同一个资源或数据的情况。
Python标准模块--concurrent.futures
线程池
子主题
0 条评论
下一页