进程&＃xff08;Process&＃xff09;
1&＃xff09;进程就是正在运行的程序,它是操作系统中,资源分配的最小单位
&＃xff08;1&＃xff09;资源分配:分配的是cpu和内存等物理资源
&＃xff08;2&＃xff09;进程号是进程的唯一标识
2&＃xff09;同一个程序执行两次之后是两个进程
3&＃xff09;进程和进程之间的关系: 数据彼此隔离,通过socket通信
获取进程id
import os
res &＃61; os.getpid() # 获取当前进行id &＃xff08;当前子进程&＃xff09;
print(res)
res &＃61; os.getppid() # 获取父进程id
print(res)
# result
"""
13724
13800
"""
进程的基本用法
from multiprocessing import Process
import time
import os
def func():
print("S2>>>当前子进程id:%s,它的父进程id:%s" %(os.getpid(),os.getppid()))
if __name__ &＃61;&＃61; "__main__":
print("S1>>>子进程id&＃xff1a;%s&＃xff0c;父进程id&＃xff1a;%s"%(os.getpid(),os.getppid()))
p &＃61; Process(target&＃61;func) # 创建子进程 target &＃61; 函数 单独用一个进程去执行谁,去完成哪个任务
p.start() # 调用子进程
# result
"""
S1>>>子进程id&＃xff1a;7576&＃xff0c;父进程id&＃xff1a;13800
S2>>>当前子进程id:7100,它的父进程id:7576
"""
带有参数的函数
异步程序&＃xff1a;不等没一行代码执行完毕&＃xff0c;就往下执行其它代码
from multiprocessing import Process
import time
import os
def func():
for i in range(1,5):
print("S2>>>当前子进程id&＃xff1a;%s&＃xff0c;它的父进程id&＃xff1a;%s"%(os.getpid(),os.getppid()))
if __name__ &＃61;&＃61; "__main__":
print("S1>>>子进程id&＃xff1a;%s&＃xff0c;父进程id&＃xff1a;%s"%(os.getpid(),os.getppid()))
p &＃61; Process(target&＃61;func)
p.start()
n &＃61; 5
for i in range(1,n&＃43;1):
print("*" * i )
# result
"""
S1>>>子进程id&＃xff1a;7928&＃xff0c;父进程id&＃xff1a;13800
*
**
***
****
*****
S2>>>当前子进程id&＃xff1a;8112&＃xff0c;它的父进程id&＃xff1a;7928
S2>>>当前子进程id&＃xff1a;8112&＃xff0c;它的父进程id&＃xff1a;7928
S2>>>当前子进程id&＃xff1a;8112&＃xff0c;它的父进程id&＃xff1a;7928
S2>>>当前子进程id&＃xff1a;8112&＃xff0c;它的父进程id&＃xff1a;7928
"""
进程之间数据&＃xff0c;彼此隔离
from multiprocessing import Process
import time
import os
count &＃61; 99
def func():
global count
count &＃43;&＃61; 1
print("当前子进程id:%s"%(os.getpid()),count)
if __name__ &＃61;&＃61; "__main__":
p &＃61; Process(target&＃61;func)
p.start()
time.sleep(1) # 让子进程跑完&＃xff0c;看是否通过子进程修改
print("主进程",count)
# result
"""
当前子进程id:1956 100
主进程 99
"""
多进程之间的并发
多程序并发时&＃xff0c;因为cpu的调度策略&＃xff0c;任务的执行是互相抢占cpu资源的过程
from multiprocessing import Process
import time
import os
def func(args):
print("S2>>>当前子进程id:%s&＃xff0c;它的父进程id&＃xff1a;%s"%(os.getpid(),os.getppid()))
print("end>>>",args)
if __name__ &＃61;&＃61; "__main__":
for i in range(10):
Process(target&＃61;func,args&＃61;(i,)).start()
print("主进程执行结束...")
# result
"""
主进程执行结束...
S2>>>当前子进程id:10376&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 2
S2>>>当前子进程id:13476&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 1
S2>>>当前子进程id:11812&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 3
S2>>>当前子进程id:13936&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 4
S2>>>当前子进程id:2384&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 7
S2>>>当前子进程id:7024&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 5
S2>>>当前子进程id:14284&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 8
S2>>>当前子进程id:4816&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 9
S2>>>当前子进程id:120&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 6
S2>>>当前子进程id:8604&＃xff0c;它的父进程id&＃xff1a;9932
end>>> 0
"""
主进程和子进程之间的关系
&＃xff08;1&＃xff09;主进程执行完所有代码&＃xff0c;等待子进程全部结束&＃xff0c;再终止程序
&＃xff08;2&＃xff09;不等待&＃xff0c;子进程变成僵尸用户&＃xff0c;在后台不停运行&＃xff0c;耗费资源
from multiprocessing import Process
import time
import os
def func(args):
print("S2>>>当前子进程id&＃xff1a;%s&＃xff0c;它的父进程id&＃xff1a;%s"%(os.getpid(),os.getppid()))
time.sleep(0.1)
print("end>>>",args)
if __name__ &＃61;&＃61; "__main__":
for i in range(10):
Process(target&＃61;func,args&＃61;(i,)).start()
print("主进程执行结束...")
# result
"""
主进程执行结束...
S2>>>当前子进程id&＃xff1a;10092&＃xff0c;它的父进程id&＃xff1a;6304
S2>>>当前子进程id&＃xff1a;11220&＃xff0c;它的父进程id&＃xff1a;6304
S2>>>当前子进程id&＃xff1a;12992&＃xff0c;它的父进程id&＃xff1a;6304
S2>>>当前子进程id&＃xff1a;8384&＃xff0c;它的父进程id&＃xff1a;6304
S2>>>当前子进程id&＃xff1a;3824&＃xff0c;它的父进程id&＃xff1a;6304
S2>>>当前子进程id&＃xff1a;12880&＃xff0c;它的父进程id&＃xff1a;6304
S2>>>当前子进程id&＃xff1a;6352&＃xff0c;它的父进程id&＃xff1a;6304
end>>> 0
end>>> 2
S2>>>当前子进程id&＃xff1a;7160&＃xff0c;它的父进程id&＃xff1a;6304
end>>> 9
end>>> 7
S2>>>当前子进程id&＃xff1a;6676&＃xff0c;它的父进程id&＃xff1a;6304
end>>> 5
end>>> 8
end>>> 3
S2>>>当前子进程id&＃xff1a;7488&＃xff0c;它的父进程id&＃xff1a;6304
end>>> 1
end>>> 6
end>>> 4
"""
join
等待子进程执行完毕之后&＃xff0c;主进程再向下执行
基本用法
from multiprocessing import Process
import time
import os
def func():
print("发送第一封邮件")
if __name__ &＃61;&＃61; "__main__":
p &＃61; Process(target&＃61;func)
p.start()
time.sleep(1)
p.join() # 针对于p进程对象来说,必须等待p这个进程任务执行完毕之后,主进程的代码在向下执行
print("发送第二封邮件")
# result
"""
发送第一封邮件
发送第二封邮件
"""
多个子进程
多个子进程通过join加阻塞&＃xff0c;可以实现同步控制
from multiprocessing import Process
import time
import os
def func(index):
print("第%s封邮件发送..."%(index))
if __name__ &＃61;&＃61; "__main__":
lst &＃61; [] # 初始化一个列表
for i in range(5):
p &＃61; Process(target&＃61;func,args&＃61;(i,))
p.start()
lst.append(p)
p.join()
# 循环列表中每一个进程&＃xff0c;都加上join&＃xff0c;保证每个子进程执行完毕&＃xff0c;保证同步性
for i in lst:
i.join()
print("发送最后一封邮件...")
# result
"""
第0封邮件发送...
第1封邮件发送...
第2封邮件发送...
第3封邮件发送...
第4封邮件发送...
发送最后一封邮件...
"""
自定义类方式创建进程
基本用法
from multiprocessing import Process
import time
import os
class MyProcess(Process): # 必须继承父类Process
def run(self): # 必须写成run方法
print(">>>子进程id&＃xff1a;%s&＃xff0c;父进程id&＃xff1a;%s"%(os.getpid(),os.getppid()))
if __name__ &＃61;&＃61; "__main__":
p &＃61; MyProcess()
p.start()
print(">>>主进程id&＃xff1a;%s"%(os.getpid()))
# result
"""
>>>主进程id&＃xff1a;9988
>>>子进程id&＃xff1a;14248&＃xff0c;父进程id&＃xff1a;9988
"""
带参数子进程函数
from multiprocessing import Process
import time
import os
class MyProcee(Process):
def __init__(self,arg):
super().__init__() # 必须调用一下父类的构造方法
self.arg &＃61; arg
def run(self):
print(">>>子进程id&＃xff1a;%s&＃xff0c;父进程id&＃xff1a;%s"%(os.getpid(),os.getppid()))
print(self.arg)
if __name__ &＃61;&＃61; "__main__":
lst &＃61; []
for i in range(10):
p &＃61; MyProcee("参数&＃xff1a;%s"%(i))
p.start()
lst.append(p)
for i in lst:
i.join()
print("最后打印子进程id",os.getppid())
# result
"""
>>>子进程id&＃xff1a;7488&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;6
>>>子进程id&＃xff1a;11300&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;8
>>>子进程id&＃xff1a;12048&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;4
>>>子进程id&＃xff1a;7868&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;0
>>>子进程id&＃xff1a;4792&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;5
>>>子进程id&＃xff1a;10764&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;2
>>>子进程id&＃xff1a;10852&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;1
>>>子进程id&＃xff1a;9348&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;3
>>>子进程id&＃xff1a;9860&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;7
>>>子进程id&＃xff1a;8956&＃xff0c;父进程id&＃xff1a;8596
参数&＃xff1a;9
最后打印子进程id 13800
"""
并行和并发
&＃xff08;1&＃xff09;并行&＃xff1a;一个cpu同一时间不停执行多个程序
&＃xff08;2&＃xff09;并发&＃xff1a;多个cpu同一时间不停执行多个程序
cpu的进程调度方法
&＃xff08;1&＃xff09;先来先服务fcfs(first come first server):先来的先执行
&＃xff08;2&＃xff09;短作业优先算法:分配的cpu多,先把短的算完
&＃xff08;3&＃xff09;时间片轮转算法:每一个任务就执行一个时间片的时间.然后就执行其他的
&＃xff08;4&＃xff09;多级反馈队列算法
越是时间长的,cpu分配的资源越短,优先级靠后&＃xff0c;越是时间短的,cpu分配的资源越多
进程三状态图
&＃xff08;1&＃xff09;就绪(Ready)状态-->只剩下CPU需要执行外,其他所有资源都已分配完毕 称为就绪状态。
&＃xff08;2&＃xff09;执行(Running)状态 --> cpu开始执行该进程时称为执行状态
&＃xff08;3&＃xff09;阻塞(Blocked)状态 --> 由于等待某个事件发生而无法执行时,便是阻塞状态,cpu执行其他进程.例如&＃xff0c;等待I/O完成input、申请缓冲区不能满足等
同步|异步/阻塞|非阻塞
场景在多任务中
同步&＃xff1a;必须等我这件事干完了,你在干,只有一条主线,就是同步
异步&＃xff1a;没等我这件事情干完,你就在干了,有两条主线,就是异步
阻塞&＃xff1a;比如代码有了input,就是阻塞,必须要输入一个字符串,否则代码不往下执行
非阻塞&＃xff1a;没有任何等待,正常代码往下执行
同步阻塞&＃xff1a;效率低,cpu利用不充分
异步阻塞&＃xff1a;比如socketserver,可以同时连接多个,但是彼此都有recv
同步非阻塞&＃xff1a;没有类似input的代码,从上到下执行.默认的正常情况代码
异步非阻塞&＃xff1a;效率是最高的,cpu过度充分,过度发热
守护进程
可以给子进程贴上守护进程的名字,该进程会随着主进程代码执行完毕而结束(为主进程守护)
&＃xff08;1&＃xff09;守护进程会在主进程代码执行结束后就终止
&＃xff08;2&＃xff09;守护进程内无法再开启子进程,否则抛出异常(了解)
基本用法
from multiprocessing import Process
import time
import os
def func():
print("子进程start...")
time.sleep(0.1)
print("子进程end...")
if __name__ &＃61;&＃61; "__main__":
p &＃61; Process(target&＃61;func)
p.daemon &＃61; True # 在start开始前设置该进程为守护进程
p.start()
print("主进程执行结束...")
# result
"""
主进程执行结束...
"""
多个子进程
&＃xff08;1&＃xff09;当主进程里面的代码全部执行完毕&＃xff0c;守护进程自动终止
&＃xff08;2&＃xff09;func1是守护进程&＃xff0c;func2是非守护进程
from multiprocessing import Process
import time
import os
def func1():
count &＃61; 1
while True:
print("*" * count)
time.sleep(0.5)
count &＃43;&＃61; 1
def func2():
print("func2 start...")
time.sleep(2)
print("func2 end...")
if __name__ &＃61;&＃61; "__main__":
p1 &＃61; Process(target&＃61;func1)
p1.daemon &＃61; True
p2 &＃61; Process(target&＃61;func2)
p1.start()
p2.start()
time.sleep(1)
print("主进程代码执行完毕...")
# result
"""
func2 start...
*
**
主进程代码执行完毕...
func2 end...
"""
守护进程应用--报活
from multiprocessing import Process
import time
import os
def alive():
while True:
print("一号服务主机ok...")
time.sleep(1) # 相隔1秒报活
def func():
print("一号服务主机负责统计日志信息")
time.sleep(3)
if __name__ &＃61;&＃61; "__main__":
p1 &＃61; Process(target&＃61;alive)
p1.daemon &＃61; True
p1.start()
p2 &＃61; Process(target&＃61;func)
p2.start()
p2.join() # join添加阻塞&＃xff0c;join执行结束&＃xff0c;代表服务统计功能失败
print(" ... ")
# result
"""
一号服务主机ok...
一号服务主机负责统计日志信息
一号服务主机ok...
一号服务主机ok...
一号服务主机ok...
...
"""
锁&＃xff08;LOCK&＃xff09;
&＃xff08;1&＃xff09;lock.acquire() 上锁
&＃xff08;2&＃xff09;lock.release() 解锁
&＃xff08;3&＃xff09;同一时间允许一个进程上一把锁&＃xff0c;就是lock。加锁可以保证多个进程修改同一块数据&＃xff0c;同一时间只能有一个任务进行修改&＃xff0c;即串行修改&＃xff0c;速度慢&＃xff0c;保证安全。
&＃xff08;4&＃xff09;同一时间允许多个进程上多把锁&＃xff0c;就是信号量[Semaphore]
信号量是锁的变形&＃xff1a;实现是 计数器 &＃43; 锁&＃xff0c;同时允许多个进程上锁
&＃xff08;5&＃xff09;互斥锁Lock&＃xff1a;就是进程的互相排斥&＃xff0c;谁先抢到资源&＃xff0c;谁就上锁修改资源内容&＃xff0c;保证数据的同步性
多个锁一起上&＃xff0c;不开锁&＃xff0c;会造成死锁&＃xff0c;上锁和解锁是一对
基本语法
import json,time
lock &＃61; Lock() # 创建一把锁
lock.acquire() # 上锁
print(123)
lock.release() # 解锁
死锁
程序添加了阻塞&＃xff0c;程序不能往下执行。下面代码程序阻塞&＃xff0c;也不会打印111
lock &＃61; Lock() # 创建一把锁
lock.acquire() # 上锁
lock.acquire()
print(111)
lock.release() # 解锁
抢票示例
from multiprocessing import Lock,Process
import time
count &＃61; 1
def get_ticket(person):
global count
time.sleep(0.1)
if count > 0:
print("%s抢到票了..."%(person))
count -&＃61; 1
else:
print("%s没有抢到票..."%(person))
# 定义一个函数来进行统一调用
def ticket(person,lock):
print("%s 查询余票&＃xff1a;%s"%(person,count)) # 查询余票
lock.acquire()
get_ticket(person)
lock.acquire()
if __name__ &＃61;&＃61; "__main__":
lock &＃61; Lock()
for i in range(10):
p &＃61; Process(target&＃61;ticket,args&＃61;("person%s"%(i),lock))
p.start()
# result
"""
person6 查询余票&＃xff1a;1
person0 查询余票&＃xff1a;1
person1 查询余票&＃xff1a;1
person6抢到票了...
person4 查询余票&＃xff1a;1
person9 查询余票&＃xff1a;1
person3 查询余票&＃xff1a;1
person8 查询余票&＃xff1a;1
person2 查询余票&＃xff1a;1
person7 查询余票&＃xff1a;1
person5 查询余票&＃xff1a;1
"""
信号量
本质上就是锁&＃xff0c;只不过可以控制锁的数量
from multiprocessing import Process, Semaphore
import os, time
def ktv(person, sem):
sem.acquire()
print("%s进入ktv开始唱歌" % (person))
print(os.getpid())
time.sleep(3)
print("%s走出ktv离开歌房" % (person))
sem.release()
if __name__ &＃61;&＃61; "__main__":
sem &＃61; Semaphore(1)
for i in range(5):
p &＃61; Process(target&＃61;ktv, args&＃61;("person%s" % (i), sem))
p.start()
# result
"""
person3进入ktv开始唱歌
15956
person3走出ktv离开歌房
person2进入ktv开始唱歌
14632
person2走出ktv离开歌房
person4进入ktv开始唱歌
9248
person4走出ktv离开歌房
person1进入ktv开始唱歌
8816
person1走出ktv离开歌房
person0进入ktv开始唱歌
12992
person0走出ktv离开歌房
"""
事件&＃xff08;Event&＃xff09;
&＃xff08;1&＃xff09;阻塞事件
e &＃61; Event() 生成事件对象e
e.wait() 动态给程序加阻塞&＃xff0c;程序当中是否加阻塞完全取决于该对象中的is_set() [默认返回值是False]&＃xff0c;True不加阻塞
&＃xff08;2&＃xff09;控制这个属性的值
set() 方法&＃xff1a;将这个属性的值改成True
clear() 方法&＃xff1a;将这个属性的值改成False
is_set() 方法&＃xff1a;判断当前的属性是否为True (默认上来是False)
基本语法
from multiprocessing import Process,Event
import time,random
e &＃61; Event()
print(e.is_set())
e.wait(1) # 最多阻塞时间为1秒
print(111)
# result
"""
False
111
"""
from multiprocessing import Process,Event
import time,random
e &＃61; Event()
e.set() # 将内部属性改成True
e.wait()
print(111)
e.clear() # 将内部属性改成False
e.wait()
print(222)
# result
"""
111
"""
模拟交通灯
&＃xff08;1&＃xff09;两个信号灯会交替变换
from multiprocessing import Process,Event
import time,random
def traffic_light(e):
print("红灯亮")
while True:
if e.is_set():
time.sleep(1) # 当前绿灯&＃xff0c;等待1秒
print("红灯亮") # 等完1秒后&＃xff0c;变成红灯
e.clear()
else:
time.sleep(1) # 当前是红灯
print("绿灯亮")
e.set()
e &＃61; Event()
traffic_light(e)
&＃xff08;2&＃xff09;模拟车遇到红灯停&＃xff0c;绿灯行的操作
from multiprocessing import Process,Event
import time,random
def traffic_light(e):
print("红灯亮")
while True:
if e.is_set():
time.sleep(1) # 当前绿灯&＃xff0c;等待1秒
print("红灯亮") # 等完1秒后&＃xff0c;变成红灯
e.clear()
else:
time.sleep(1) # 当前是红灯
print("绿灯亮")
e.set()
def car(e,i):
if not e.is_set():
print("car%s在等待..."%(i))
e.wait()
print("car%s通行..."%(i))
if __name__ &＃61;&＃61; "__main__":
e &＃61; Event()
lst &＃61; []
p1 &＃61; Process(target&＃61;traffic_light,args&＃61;(e,))
p1.daemon &＃61; True # 设置守护进程&＃xff0c;小车跑完&＃xff0c;交通灯终止
p1.start()
# 模拟通行
for i in range(5):
time.sleep(random.uniform(0,2)) # 随机停0-2秒
p2 &＃61; Process(target&＃61;car,args&＃61;(e,i))
p2.start()
lst.append(p2) # 将进程添加到列表中
for i in lst:
i.join() # 等所有车跑完&＃xff0c;终止交通灯&＃xff1b;加一个等待
print("程序结束...")
# result
"""
红灯亮
绿灯亮
car0通行...
程序结束...
红灯亮
car1在等待...
绿灯亮
car1通行...
程序结束...
car2通行...
程序结束...
红灯亮
car3在等待...
绿灯亮
car3通行...
程序结束...
car4通行...
程序结束...
"""
进程间通信IPC
&＃xff08;1&＃xff09;IPC Inter-Process Communication
&＃xff08;2&＃xff09;进程之间通信的两种机制&＃xff1a;管道Pipe 队列 Queue
&＃xff08;3&＃xff09;put() 存放
&＃xff08;4&＃xff09;get() 获取
&＃xff08;5&＃xff09;get_nowait() 拿不到报异常
&＃xff08;6&＃xff09;put_nowait() 非阻塞版本的put
&＃xff08;7&＃xff09;q.empty() 检测是否为空
&＃xff08;8&＃xff09;full() 检测是否已经存满
基本语法
&＃xff08;1&＃xff09;先进先出&＃xff0c;后进后出
&＃xff08;2&＃xff09;让进程之间形成数据之间的共享
from multiprocessing import Process,Queue
q &＃61; Queue()
q.put(1) # 把数据放到q队列中
res &＃61; q.get() # 把数据从队列中取出
print(res)
res &＃61; q.get() # 没有数据后会形成阻塞
res &＃61; q.get_nowait() # 没有数据&＃xff0c;直接报错
queue指定队列长度
from multiprocessing import Queue
q &＃61; Queue(3)
q.put(1)
q.put(2)
q.put(3)
# q.put(4) # 存放值满了&＃xff0c;阻塞
q.put_nowait(5) # 往队列中存值&＃xff0c;满直接报错
进程通信&＃xff0c;依赖Queue
from multiprocessing import Queue,Process
# 子进程
def func(q):
res &＃61; q.get() # 子进程获取数据
print(res)
q.put(1) # 子进程添加数据
if __name__ &＃61;&＃61; "__main__":
q &＃61; Queue()
p &＃61; Process(target&＃61;func,args&＃61;(q,))
p.start()
q.put(2) # 主进程添加数据
p.join()
res &＃61; q.get() # 主进程添加数据
print(res)
print("程序结束...")
# result
"""
2
1
程序结束...
"""
生产者与消费者模型
如&＃xff1a;1号进程负责爬取网页中内容&＃xff0c;2号进程负责匹配提取网页中关键字
1号进程可以看成一个生产者&＃xff0c;2号进程可以看成一个消费者
from multiprocessing import Process,Queue
import os,random,time
# 消费者模型&＃xff08;负责取值&＃xff09;
def consumer(q,name):
while True:
food &＃61; q.get() # 拿出数据
if food is None:
break
time.sleep(random.uniform(0.1,1))
print("%s 吃了一个 %s"%(name,food))
# 生产者模型&＃xff08;负责存值&＃xff09;
def producer(q,name,food):
for i in range(3):
time.sleep(random.uniform(0.1,1))
print("%s 生产了 %s&＃xff0c;%s"%(name,food,i))
q.put(food &＃43; str(i))
if __name__ &＃61;&＃61; "__main__":
q &＃61; Queue()
c1 &＃61; Process(target&＃61;consumer,args&＃61;(q,"A"))
c1.start()
p1 &＃61; Process(target&＃61;producer,args&＃61;(q,"B","egg"))
p1.start()
p1.join()
q.put(None)
# result
"""
B 生产了 egg&＃xff0c;0
A 吃了一个 egg0
B 生产了 egg&＃xff0c;1
A 吃了一个 egg1
B 生产了 egg&＃xff0c;2
A 吃了一个 egg2
"""
进程池
&＃xff08;1&＃xff09;开启过多的进程并不一定提高效率
cpu负载任务过多&＃xff0c;平均单个任务执行效率低&＃xff0c;较低执行速度
&＃xff08;2&＃xff09;apply 开启进程&＃xff0c;同步阻塞&＃xff0c;每次都要等待当前任务完成之后&＃xff0c;开启下一个进程
&＃xff08;3&＃xff09;apply_async 开启进程&＃xff0c;异步非阻塞&＃xff08;主进程和子进程异步&＃xff09;
线程
&＃xff08;1&＃xff09;进程是资源分配的最小单位&＃xff0c;线程是计算机中调度的最小单位
&＃xff08;2&＃xff09;线程的缘起
资源分配需要分配内存空间&＃xff0c;分配cpu&＃xff0c;分配的内存空间存放着临时要处理的数据&＃xff0c;如要执行的代码&＃xff0c;数据。
&＃xff08;3&＃xff09;线程的特点
轻量级&＃xff0c;能干更多的活&＃xff0c;一个进程中所有线程资源共享&＃xff0c;一个进程至少一个线程工作
一个进程可以有多个线程&＃xff0c;共享一份资源
from threading import Thread
from multiprocessing import Process
import os,time,random
def func(num):
time.sleep(random.uniform(0.1,1))
print("子线程",num,os.getpid())
if __name__ &＃61;&＃61; "__main__":
for i in range(5):
t &＃61; Thread(target&＃61;func,args&＃61;(i,))
t.start()
# result
"""
子线程 1 6348
子线程 3 6348
子线程 0 6348
子线程 4 6348
子线程 2 6348
"""
并发多进程线程对比
from threading import Thread
from multiprocessing import Process
import os,time,random
# 计算多线程时间
def func(i):
# print("子线程",i,os.getpid())
pass
if __name__ &＃61;&＃61; "__main__":
lst &＃61; []
start_time &＃61; time.time()
for i in range(100):
t &＃61; Thread(target&＃61;func,args&＃61;(i,))
t.start()
lst.append(t)
for i in lst:
i.join()
end_time &＃61; time.time()
print("time:",end_time-start_time)
# result
"""
time: 0.02393651008605957
"""
from threading import Thread
from multiprocessing import Process
import os,time,random
def func(i):
# print("子线程",i,os.getpid())
pass
if __name__ &＃61;&＃61; "__main__":
lst &＃61; []
start_time &＃61; time.time()
for i in range(100):
p &＃61; Process(target&＃61;func,args&＃61;(i,))
p.start()
lst.append(p)
for i in lst:
i.join()
end_time &＃61; time.time()
print("time:",end_time-start_time)
# result
"""
time: 13.565200328826904
"""
多线程共享同一份进程资源
from threading import Thread
from multiprocessing import Process
import os,time,random
num &＃61; 100
lst &＃61; []
def func(i):
global num
num -&＃61; 1
for i in range(100):
t &＃61; Thread(target&＃61;func,args&＃61;(i,))
t.start()
lst.append(t)
for i in lst:
i.join()
print(num)
# result
"""
0
"""
线程相关函数
&＃xff08;1&＃xff09;线程.is_alive() 检测线程是否仍然存在
&＃xff08;2&＃xff09;线程.setName() 设置线程名字
&＃xff08;3&＃xff09;线程。getName() 获取线程名字
&＃xff08;4&＃xff09;currentThread().ident 查看线程id
&＃xff08;5&＃xff09;enumerate() 返回目前正在运行的线程列表
&＃xff08;6&＃xff09;activeCount() 返回目前正在运行的线程数量
查看线程id
from threading import currentThread,Thread
import os
def func():
print("子线程",currentThread().ident)
t &＃61; Thread(target&＃61;func)
t.start()
print("主线程",currentThread().ident,os.getpid())
# result
"""
子线程 9952
主线程 912 13268
"""
enumerate()
from threading import currentThread,Thread,enumerate
import os,time
def func():
print("子线程",currentThread().ident)
time.sleep(0.5)
for i in range(5):
t &＃61; Thread(target&＃61;func)
t.start()
print(enumerate())
print(len(enumerate()))
# result
"""
子线程 3668
子线程 5856
子线程 13288
子线程 6304
子线程 5496
[<_MainThread(MainThread, started 8644)>,
6
activeCount()
from threading import currentThread,Thread,enumerate,activeCount
import os,time
def func():
print("子线程",currentThread().ident)
time.sleep(1)
for i in range(5):
Thread(target&＃61;func).start()
print(activeCount())
# result
"""
子线程 5168
子线程 3428
子线程 13808
子线程 11732
子线程 2756
6
"""
守护线程
等待所有线程执行结束后自动结束&＃xff0c;守护所有线程
from threading import currentThread,Thread,enumerate,activeCount
import os,time
def func1():
while True:
time.sleep(1)
print("线程1")
def func2():
print("线程2-start")
time.sleep(2)
print("线程2-end")
t1 &＃61; Thread(target&＃61;func1)
t1.setDaemon(True)
t1.start()
t2 &＃61; Thread(target&＃61;func2)
t2.start()
print("主线程执行结束...")
# result
"""
线程2-start
主线程执行结束...
线程1
线程2-end
"""
线程的数据安全 依赖Lock
from threading import currentThread,Thread,enumerate,activeCount,Lock
import os,time
n &＃61; 0
def func1(lock):
global n
for i in range(1000):
lock.acquire() # 上锁
n -&＃61; 1
lock.release() # 解锁
def func2(lock):
global n
for i in range(1000):
with lock:
n &＃43;&＃61; 1
if __name__ &＃61;&＃61; "__main__":
lock &＃61; Lock()
lst &＃61; []
for i in range(5):
t1 &＃61; Thread(target&＃61;func1,args&＃61;(lock,))
t2 &＃61; Thread(target&＃61;func2,args&＃61;(lock,))
t1.start()
t2.start()
lst.append(t1)
lst.append(t2)
for i in lst:
i.join()
print("主线程结束")
print(n)
# result
"""
0
"""
信号量
from threading import currentThread,Thread,enumerate,activeCount,Lock,Semaphore
import os,time,random
def func(i,sem):
time.sleep(random.uniform(0.1,1))
with sem:
print(i)
time.sleep(2)
if __name__ &＃61;&＃61; "__main__":
sem &＃61; Semaphore(5)
for i in range(5):
Thread(target&＃61;func,args&＃61;(i,sem)).start()
# result
"""
2
1
0
4
3
"""
死锁&＃xff0c;递归锁&＃xff0c;互斥锁
线程队列
协程&＃xff08;gevent&＃xff09;
&＃xff08;1&＃xff09;协程也叫纤程&＃xff0c;协程是线程的一种实现
一条线程可以在多任务之间来回切换&＃xff0c;对于cpu&＃xff0c;操作系统协程不存在
&＃xff08;2&＃xff09;协程的实现
协程记住任务执行到那个位置&＃xff0c;并且实现安全的切换
一个任务一旦阻塞&＃xff0c;立刻切换到另一个任务继续执行&＃xff0c;保证线程总是忙碌
&＃xff08;3&＃xff09;一个线程可以由多个协程实现&＃xff0c;协程之间不会产生数据安全问题
&＃xff08;4&＃xff09;协程模块
greenlet gevent的底层&＃xff0c;协程&＃xff0c;切换的模块
gevent 直接用&＃xff0c;gevent能提供更全面的功能
def gen():
for i in range(5):
yield i
mygen &＃61; gen()
for i in mygen:
print(i)
# result
"""
0
1
2
3
4
"""
协程改写生产者消费者模型
def producer():
for i in range(5):
yield i
def consumer():
g &＃61; producer()
for i in g:
print(i)
consumer()
# result
"""
0
1
2
3
4
"""
协程的具体实现
import time
from greenlet import greenlet
def eat():
print("eat one")
g2.switch()
time.sleep(3)
print("eat two")
def play():
print("play one")
time.sleep(1)
print("playtwo")
g1.switch()
g1 &＃61; greenlet(eat)
g2 &＃61; greenlet(play)
g1.switch()
# result
"""
eat one
play one
playtwo
eat two
"""
gevent不能识别阻塞
from gevent import monkey
# mokey.patch_all 可以把下面引入的所有模块中的阻塞,重新识别出来
monkey.patch_all()
import time
import gevent
def eat():
print("eat one")
time.sleep(1)
print("eat two")
def play():
print("play one")
time.sleep(1)
print("playtwo")
g1 &＃61; gevent.spawn(eat)
g2 &＃61; gevent.spawn(play)
g1.join()
g2.join()
print("主线程执行结束")
# result
"""
eat one
play one
eat two
playtwo
主线程执行结束
"""
京公网安备 11010802041100号