#Python并发
- 多任务
- 多进程
- 多线程
- 线程同步
#多任务处理
- 多任务处理：使得计算机可以同时处理多个任务
- 听歌的同时QQ聊天、办公、下载文件
- 实现方式：多进程、多线程
#程序和进程
- 程序：是一个指令的集合
- 进程：:正在执行的程序；或者说：当你运行一个程序，你就启动了一个进程
- 编写完的代码，没有运行时，称为程序，正在运行的代码，称为进程
- 程序是死的(静态的)，进程是活的(动态的)
- 操作系统轮流让各个任务交替执⾏ ，由于CPU的执⾏速度实在是太快了， 我们感觉就像所有任务都在同时执⾏⼀样
- 多进程中， 每个进程中所有数据（包括全局变量） 都各有拥有⼀份， 互不影响
#并发与并行
#并发
- 并发：单CPU，多进程并发
- 无论是并行还是并发，在用户看来都是'同时'运行的，不管是进程还是线程，都只是一个任务而已，真实干活的是cpu，cpu来做这些任务，而一个cpu同一时刻只能执行一个任务
- 并发：是伪并行，即看起来是同时运行。单个cpu+多道技术就可以实现并发，（并行也属于并发）
#并行
-
并行：多CPU（同时运行，只有具有多个cpu才能实现并行）
-
单核下，可以利用多道技术，多个核，每个核也都可以利用多道技术（多道技术是针对单核而言的）
-
一旦任务1遇到I/O就被迫中断执行，此时任务5就拿到cpu1的时间片去执行，这就是单核下的多道技术而一旦任务1的I/O结束了，操作系统会重新调用它(需知进程的调度、分配给哪个cpu运行，由操作系统说了算)，可能被分配给四个cpu中的任意一个去执行所有现代计算机经常会在同一时间做很多件事，一个用户的PC（无论是单cpu还是多cpu），都可以同时运行多个任务（一个任务可以理解为一个进程）。
#多道技术
- 内存中同时存入多道（多个）程序，cpu从一个进程快速切换到另外一个，使每个进程各自运行几十或几百毫秒，这样，虽然在某一个瞬间，一个cpu只能执行一个任务，但在1秒内，cpu却可以运行多个进程，这就给人产生了并行的错觉，即伪并发，以此来区分多处理器操作系统的真正硬件并行（多个cpu共享同一个物理内存）
#多进程
#多进程
- 最小的资源单位(内存)，
- 程序在开始运行时，首先会创建一个主进程
- 在主进程(父进程)下，我们可以创建新的进程(子进程)，子进程依赖于主进程，如果主进程结束，程序就会退出
- Python提供了非常好用的多进程包multiprocessing，借助这个包，可以轻松完成从单进程到并发执行的转换
#multiprocessing
-
multiprocessing模块提供了一个Process类来创建一个进程对象
from multiprocessing import Process num = 10 def r1(): global num num += 10 print(f"子进程1 num = {num}") >>>子进程1 num = 20 def r2(): global num num += 5 print(f"子进程2 num = {num}") >>>子进程2 num = 15 if __name__ == "__main__": p1 = Process(target=r1) p2 = Process(target=r2) p1.start() p2.start() p1.join() p2.join() print(f"主进程中 num = {num}") >>>主进程中 num = 10 -
name == "main"说明:
一个Python的文件有两种使用的方法，第一是直接作为程序执行，第二是import到其他的Python程序中被调用(模块重用)执行，因此 if name__ == 'main': 的作用就是控制这两种情况执行代码的过程，name 是内置变量，用于表示当前模块的名字，在if name == 'main': 下的代码只有在文件作为程序直接执行才会被执行，而import到其他程序中是不会被执行的
在 Windows 上，子进程会自动 import 启动它的这个文件，而在 import 的时候是会执行这些语句的。如果不加if name == "main":的话就会无限递归创建子进程所以必须把创建子进程的部分用那个 if 判断保护起来
import 的时候 name 不是 main ，就不会递归运行了
#Process
-
Process([group [, target [, name [, args [, kwargs]]]]])，
由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）
-
参数介绍
- group参数未使用，值始终为None
- target表示调用对象，即子进程要执行的任务
- args表示调用对象的位置参数元组，args=(1,2,'egon',)
- kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}
- name为子进程的名称
#Process常用方法
- p.start() 启动进程，并调用该子进程中的p.run()
- p.run()进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
- p.terminate() 强制终止进程p，不会进行任何清理操作
- p.is_alive() 如果p仍然运行，返回True.用来判断进程是否还在运行
- p.join() 主进程等待p终止，timeout是可选的超时时间
#Process常用属性
- name：当前进程实例别名， 默认为Process-N， N为从1开始递增的整
数； - pid：当前进程实例的PID值
from multiprocessing import Process
def qiang(name):
print(f"{name}抢地盘")
def zhan():
print("占东京")
if __name__ == "__main__": #main 代表主程序
# #name是内置变量，用来表示当前模块的名字
print("主进程启动")
print(__name__) >>>__main__
haonan = Process(target=qiang,args=("爸爸",),name= "进程1")
jiba = Process(target=zhan,name="进程2")
haonan.start()
jiba.start()
print(haonan.name,jiba.name) >>>进程1 进程2
print(haonan.pid,jiba.pid) >>>8772 16844
haonan.join() >>>爸爸抢地盘
jiba.join() >>>占东京
print("结束")
#继承Process类
- 创建新的进程还能够使用类的方式，可以自定义一个类，继承Process类，每次实例化这个类的时候，就等同于实例化一个进程对象
from multiprocessing import Process
class Jc(Process):
def run(self): #将父类Process的run方法重写
print("固定执行这些代码")
def paly(self):
print("qweqwe")
if __name__ == "__main__":
p1 = Jc()
p1.play()
p1.start()
p1.join()
###继承Process类重新run方法时，可以填写新的方法，但是不能重新定义init类属性
#进程池
- 进程池：用来创建多个进程
- 当需要创建的⼦进程数量不多时， 可以直接利⽤multiprocessing中的Process动态生成多个进程， 但如果是上百甚⾄上千个⽬标， ⼿动的去创建进程的⼯作量巨⼤，此时就可以⽤到multiprocessing模块提供的__Pool__
- 初始化Pool时， 可以指定⼀个最⼤进程数， 当有新的请求提交到Pool中时，如果池还没有满， 那么就会创建⼀个新的进程⽤来执⾏该请求； 但如果池中的进程数已经达到指定的最⼤值， 那么该请求就会等待， 直到池中有进程结束， 才会创建新的进程来执⾏
from multiprocessing import Pool
import time
def e1(name):
print("aasss我是e1",name)
# time.sleep(5)
def e2():
print("123123我是e2")
# time.sleep(2)
if __name__ == "__main__":
po = Pool(3) #定义一个进程池 最大进程数为几 此处为3 默认为CPU核数
for i in range(5):
po.apply_async(e1("baba")) 使⽤⾮阻塞⽅式调⽤e1
po.apply_async(e2)使⽤⾮阻塞⽅式调⽤e2
po.close()
po.join()
#Pool常⽤函数解析
- p.apply_async(func[, args[, kwds]]) ： 使⽤⾮阻塞⽅式调⽤func（并⾏执
⾏， 堵塞⽅式必须等待上⼀个进程退出才能执⾏下⼀个进程） ， args为传递给func的参数列表， kwds为传递给func的关键字参数列表； - p.apply(func[, args[, kwds]])（了解即可） 使⽤阻塞⽅式调⽤func
- p.close()： 关闭Pool， 使其不再接受新的任务；
- p.join()： 主进程阻塞， 等待⼦进程的退出， 必须在close或terminate之后使⽤；
#进程间通信(Queue)
- 多进程之间是默认不共享数据的
- 通过Queue可实现进程间的数据传递
- Q本身是一个消息队列
- 如何添加消息(如队操作)
#Queue
- 可以使用multiprocessing模块和Queue实现多进程之间的数据传递
- 初始化Queue()对象时，若括号中没有指定最大可接受的消息数量，或数量为负值， 那么就代表可接受的消息数量没有上限
- Queue.qsize()： 返回当前队列包含的消息数量
- Queue.empty()： 如果队列为空， 返回True， 反之False
- Queue.full()： 如果队列满了， 返回True,反之False
- Queue.get([block[, timeout]])： 获取队列中的⼀条消息， 然后将其从列队中移除， block默认值为True
- 如果block使⽤默认值， 且没有设置timeout（单位秒） ， 消息列队如果为空， 此时程序将被阻塞（停在读取状态） ， 直到从消息列队读到消息为⽌，如果设置了timeout， 则会等待timeout秒， 若还没读取到任何消息， 则抛出"Queue.Empty"异常
- 如果block值为False， 消息列队如果为空， 则会立刻抛出“Queue.Empty”异常
- Queue.get_nowait()： 相当Queue.get(False)
- Queue.put(item,[block[, timeout]])： 将item消息写⼊队列， block默认值为True
- 如果block使⽤默认值， 且没有设置timeout（单位秒） ， 消息列队如果已经没有空间可写⼊， 此时程序将被阻塞（停在写⼊状态） ， 直到从消息列队腾出空间为⽌， 如果设置了True和timeout， 则会等待timeout秒， 若还没空间， 则抛出"Queue.Full"异常
- 如果block值为False， 消息列队如果没有空间可写⼊， 则会⽴刻抛
出"Queue.Full"异常
- Queue.put_nowait(item)： 相当Queue.put(item, False)；
from multiprocessing import Queue,Process
import time
def w(q):
for i in range(6):
q.put(i)
print("子进程1w进程在写", i)
def r1(q):
while 1:
if not q.empty():
print("r1进程",q.get())
time.sleep(1)
def r2(q):
while 1:
if not q.empty():
print("r2进程",q.get())
time.sleep(1)
if __name__ == "__main__":
q = Queue(7) #初始化一个queue对象，指定最多可接受的消息数量
pw = Process(target=w,args=(q,))
pr1 = Process(target=r1,args=(q,))
pr2 = Process(target=r2,args=(q,))
pw.start()
pw.join()
pr1.start()
pr2.start()
pr1.join()
pr2.join()
>>>
子进程1w进程在写 0
子进程1w进程在写 1
子进程1w进程在写 2
子进程1w进程在写 3
子进程1w进程在写 4
子进程1w进程在写 5
r1进程 0
r2进程 1
r1进程 2
r2进程 3
r1进程 4
r2进程 5
- 以上实例有问题，r1跟r2都读不到完整的数据，因为在get的时候先读取会再移除，所以需要在get读取的时候，再写一份
from multiprocessing import Queue,Process
def w(q):
for i in range(6):
q.put(i)
print("子进程1w进程在写",i)
def r1(q,q2):
while 1:
a = q.get()
print("子进程r1进程在读",a)
q2.put(a)
def r2(q2):
while 1:
print("子进程r2进程在读",q2.get())
if __name__=="__main__":
q = Queue()
q2 = Queue()
pw = Process(target=w,args=(q,))
pr1 = Process(target=r1,args=(q,q2))
pr2 = Process(target=r2,args=(q2,))
pw.start()
pw.join()
pr1.start()
pr2.start()
pr1.join()
pr2.join()
#进程池通信
-
进程池创建的进程之间通信：如果要使⽤Pool创建进程， 就需要使⽤multiprocessing.Manager()中的Queue()⽽不是multiprocessing.Queue()
-
否则会得到⼀条如下的错误信息
RuntimeError: Queue objects should only be shared between processes through inheritance
#Manager,Pool
from multiprocessing import Manager,Pool
import time
def w(q):
for i in range(6):
q.put(i)
print("写",i)
def r(q):
time.sleep(2)
for i in range(q.qsize()):
print("读",q.get(i))
if __name__ == "__main__":
q = Manager().Queue()
po = Pool()
po.apply_async(w,(q,))
po.apply_async(r,(q,))
po.close()
po.join()
#多线程
#线程
- 最小的运行单位(CPU)
- 实现多任务的另一种方式
- 个进程中，也经常需要同时做多件事，就需要同时运行多个‘子任务’，这些子任务，就是线程
- 线程又被称为轻量级进程(lightweight process)，是更小的执行单元
- 一个进程可拥有多个并行的(concurrent)线程，当中每一个线程，共享当前进程的资源
- ◆一个进程中的线程共享相同的内存单元/内存地址空间→可以访问相同的变量和对象，而且它们从同一堆中分配对象→通信、数据交换、同步操作
- 由于线程间的通信是在同一地址空间上进行的，所以不需要额外的通信机制，这就使得通信更简便而且信息传递的速度也更快
#线程与进程区别
- 进程是系统进行资源分配和调度的一个独立单位
- 进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大的提高了程序的运行效率
- 一个程序至少有一个进程，一个进程至少有一个线程
- 线程是进程的一个实体，是CPU调度和分派的基本单位，他是比进程更小的能独立运行的基本单位
- 线程的划分尺度小于进程(资源比进程少)，使得多线程序的并发性高
- 线程不能够独立运行，必须依存在进程中
- 线程和进程在使用上各有优缺点：线程执行开销小，但不利于资源的管理和保护，而进程正好相反
| 区别 | 进程 | 线程 |
|---|---|---|
| 根本区别 | 作为资源分配的单位 | 调度和执行的单位 |
| 开销 | 每个进程都有独立的代码和数据空间(进程上下文),进程间的切换会有较大的开销 | 线程可以看成轻量级的进程，统一线程共享代码和数据空间，每个线程有独立的运行栈和程序技术器(Pc),线程切换的开销小 |
| 所处环境 | 在操作系统中能同时运行多个程序 | 在同一应用程序中有多个顺序流同时执行 |
| 分配内存 | 系统在运行的时候会为每个进程分配不同的内存区域 | 除了CPU外，不会为线程分配内存(线程所使用的资源是它所属的进程的资源)，线程组只能共享资源 |
| 包含关系 | 没有线程的进程是可以看做单线程的，如果一个进程内拥有多个线程，则执行过程不是一条线的，而是多条线(线程)共同完成的 | 线程是进程的一部分，所以线程有的时候被称为是轻权进程或者轻量级进程 |
- 一般来讲：我们把进程用来分配资源，线程用来具体执行(CPU调度)
#多线程
- python的thread模块是⽐较底层的模块，在各个操作系统中表现形式不同（低级模块）
- python的threading模块是对thread做了⼀些包装的， 可以更加⽅便的被使⽤（高级模块）
- thread 有一些缺点，在threading 得到了弥补，所以我们直接学习threading
#多线程
import threading
def run1(num):
print(threading.current_thread().name)
print(num)
if __name__ == "__main__":
t1 = threading.Thread(target=run1,name = "xinc1",args=(5,)) #指定线程执行的代码
t2 = threading.Thread(target=run1,args=(10,))
t1.start() #>>>xinc1 5
t2.start() #>>>Thread-1 10 #线程名字默认为Thread-序号
- 任意一个进程默认会启动一个线程，这个线程称为主线程，主线程可以启用新的子线程
#查看当前线程数量
import threading
import time
def run(i):
print(f"线程名{threading.current_thread().name},输出：{i}")
time.sleep(2)
def play():
print(f"线程名{threading.current_thread().name}")
time.sleep(2)
if __name__ == "__main__":
for i in range(0,3):
t = threading.Thread(target=run,args=(i,))
t1 = threading.Thread(target=play)
t.start()
t1.start()
print("the end")
>>>
线程名Thread-1,输出：0
线程名Thread-2
线程名Thread-3,输出：1
线程名Thread-4
线程名Thread-5,输出：2
线程名Thread-6
the end
#创建线程两种方式：
-
第一：通过 threading.Thread 直接在线程中运行函数；
-
第二：通过继承 threading.Thread 类来创建线程
- 这种方法只需要重载 threading.Thread 类的 run 方法，然后调用 start()开启线程就可以了
import threading class my(threading.Thread): def run(self): print(f"继承threading.Thread来创建线程{self.name}") def play(self): print("这是新建的方法") if __name__ == "__main__": t1 = my() t2 = my() t1.start() t2.start() t1.play() t2.play() >>> 继承threading.Thread来创建线程Thread-1 继承threading.Thread来创建线程Thread-2 这是新建的方法 这是新建的方法
#线程的五种状态
- 多线程程序的执⾏顺序是不确定的（操作系统决定）。 当执⾏到sleep语句时， 线程将被阻塞（Blocked） ， 到sleep结束后， 线程进⼊就绪（Runnable） 状态，等待调度。 ⽽线程调度将⾃⾏选择⼀个线程执⾏。 代码中只能保证每个线程都运⾏完整个run函数， 但是线程的启动顺序、run函数中每次循环的执⾏顺序都不能确定
#线程五种状态
- 新状态：线程对象已经创建，还没有在其上调用start()方法
- 可运行状态：当线程有资格运行，但调度程序还没有把它选定为运行线程时线程所处的状态。当start()方法调用时，线程首先进入可运行状态。在线程运行之后或者从阻塞、等待或睡眠状态回来后，也返回到可运行状态
- 运行状态：线程调度程序从可运行池中选择一个线程作为当前线程时线程所处的状态。这也是线程进入运行状态的唯一一种方式。
- 等待/阻塞/睡眠状态：这是线程有资格运行时它所处的状态。实际上这个三状态组合为一种，其共同点是：线程仍旧是活的（可运行的），但是当前没有条件运行。但是如果某件事件出现，他可能返回到可运行状态
- 死亡态：当线程的run()方法完成时就认为它死去。这个线程对象也许是活的，但是，它已经不是一个单独执行的线程。线程一旦死亡，就不能复生。如果在一个死去的线程上调用start()方法，会抛出异常
#线程共享全局变量
- 在一个进程内的所有线程共享全局变量，多线程之间的数据共享(这点是比进程好)
- 缺点就是，可能造成多个线程同时修改同一个变量(即线程非安全)，可能造成混乱
#共享变量(循环次数少)
import threading
num = 0
def run1():
global num
for i in range(1000):
num += 1
print("run1",num)
def run2():
global num
for i in range(1000):
num += 1
print("run2",num)
t1 = threading.Thread(target=run1)
t2 = threading.Thread(target=run2)
t1.start()
t2.start()
print(num)
>>>
run1 1000
run2 2000
2000
#共享变量(循环次数多)
-
循环次数多的bug
import threading import time num = 0 def run1(): global num for i in range(10000000): num += 1 print("run1",num) time.sleep(2) def run2(): global num for i in range(10000000): num += 1 print("run2",num) time.sleep(2) t1 = threading.Thread(target=run1) t2 = threading.Thread(target=run2) t1.start() t2.start() time.sleep(4) print(num) >>>> run2 11930287 run1 12021610 12021610
#线程同步
#互斥锁
-
当多个线程⼏乎同时修改某⼀个共享数据的时候， 需要进⾏同步控制
-
线程同步能够保证多个线程安全访问竞争资源， 最简单的同步机制是引⼊互斥锁
-
互斥锁保证了每次只有⼀个线程进⾏写⼊操作，从⽽保证了多线程情况下数据的正确性（原子性）
-
互斥锁为资源引入一个状态：锁定/非锁定。某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；直到该线程释放资源，将资源的状态变成“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。
-
threading模块中定义了Lock类， 可以⽅便的处理锁定
lock1 = threading.Lock() ##创建锁 lock1.acquire() ##锁定 lock1.release() ##解锁
#互斥锁实例
import threading
import time
num = 0
def run1():
global num
lock.acquire()
for i in range(1000000):
num += 1
lock.release()
print("run1",num)
def run2():
global num
lock.acquire()
for i in range(1000000):
num += 1
lock.release()
print("run2",num)
lock = threading.Lock()
pe1 = threading.Thread(target=run1)
pe2 = threading.Thread(target=run2)
pe1.start()
pe2.start()
time.sleep(4)
print(num)
>>>>
run1 1000000
run2 2000000
2000000
-
当一个线程调用Lock对象的acquire()方法获得到锁时，这把锁就会进入Locked状态，因为每次只有一个线程可以获得锁，所以如果此时另一个线程2试图获得这个锁时，该线程2就会变为同步阻塞状态
直到拥有锁的线程1调用锁的release()方法释放锁之后，改锁进入unlocked状态，线程调度程序继续从处于同步阻塞状态的线程中随机选择一个来获得锁，并使得该线程进入运行状态。
-
一个线程有锁时，其他线程只能在外等待
#死锁
- 在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，这样就会造成死锁
-
死锁一：
import threading def run1(): for i in range(100): lock.acquire() print("这就是死锁") #执行完之后因为锁没释放，就不会再执行 lock = threading.Lock() t = threading.Thread(target=run1) t.start() -
死锁二：
import threading import time class my_run1(threading.Thread): def run(self): if lock1.acquire(): print("线程1锁定") time.sleep(2) if lock2.acquire(): print("线程1释放") lock2.release() lock1.release() class my_run2(threading.Thread): def run(self): if lock2.acquire(): print("线程2锁定") time.sleep(2) if lock1.acquire(): print("线程2释放") lock1.release() lock2.release() if __name__ == "__main__": lock1 = threading.Lock() lock2 = threading.Lock() t1 = my_run1() t2 = my_run2() t1.start() t2.start()在多线程程序中，死锁问题很大一部分是由线程同时获得多个锁造成的，如果一个线程获得了第一个锁，然后在获取第二个锁的时候会发生阻塞，那么这个线程就可能对其他线程造成阻塞，从而导致整个程序假死
#信号量
-
semaphore（噻么发）：用于控制一个时间点内线程进入数量的锁，信号量是用来控制线程并发数的
-
场景举例：在读写文件的时候，一般只有一个线程在写，而读可以有多个线程同时进行，如果需要线程同时读文件的线程个数，这个时候就可以使用信号量了(如果用互斥锁，就是限制同一时刻只能有一个线程读取文件)
import time import threading def run(): time.sleep(2) print("ok,") for i in range(10): t1 = threading.Thread(target = run) t1.start ######此时是无法控制同时进入的线程数 ############################################# import time import threading s1 = threading.Semaphore(5) def run(): s1.acquire() print("信号量") time.sleep(2) s1.release() for i in range(20): t = threading.Thread(target=run) t.start() ####此时控制了进入的线程数
#GIL全局解释器锁
- Cpython独有的锁，牺牲效率保障数据安全
- GIL锁是一把双刃剑，它带来优势的同时也带来一些问题
- 首先：执行Python文件是什么过程？谁把进程起来的？
- 操作系统将你的应用程序从硬盘加载到内存。运行python文件，在内存中开辟一个进程空间，将你的Python解释器以及py文件加载进去，解释器运行py文件
- Python解释器分为两部分，先将你的代码通过编译器编译成C的字节码，然后你的虚拟机拿到你的C的字节码，输出机器码，再配合操作系统把你的这个机器码扔给cpu去执行
- 你的py文件中有一个主线程，主线程做的就是这个过程。如果开多线程，每个线程都要进行这个过
- 理想情况：
- 三个线程，得到三个机器码，然后交由CPU，三个线程同时扔给三个CPU，然后同时进行，最大限度的提高效率，但是CPython多线程应用不了多核
- CPython到底干了一件什么事情导致用不了多核？
- Cpython在所有线程进入解释器之前加了一个全局解释器锁（GIL锁）。这个锁是互斥锁，是加在解释器上的，导致同一时间只有一个线程在执行所以你用不了多核
- 为什么这么干？
- 之前写python的人只有一个cpu。。。
- 所以加了一个锁，保证了数据的安全，而且在写python解释器时，更加好写了
- 为什么不取消这个锁？
- 解释器内部的管理全部是针对单线程写的，取消的话工作量很大
- 能不能不用Cpython？
- 官方推荐Cpython，处理速度快，相对其他解释器较完善。其他解释器比如pypy
- 那我该怎么办？
- 虽然多线程无法应用多核，但是多进程可以应用多核（开销大）
- Python已经有一个GIL来保证同一时间只能有一个线程来执行了，为什么我还要学互斥锁?
- 共识：锁的目的就是为了保护共享的数据，同一时间只能有一个线程来修改共享的数据
- 保护不同的数据就应该加不同的锁
- GIL和Lock是两把锁，保护的数据不一样，前者是解释器级别的(保护的就是解释器级别的数据，比如垃圾回收的数据)，后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理
#线程同步和异步
#同步调用：
- 确定调用的顺序
- 提交一个任务，自任务开始运行直到此任务结束，我再提交下一次任务
- 就是按照顺序来购买一身衣服(先买衣服，再买裤子，顺序不能反)
#异步调用：
- 不确定顺序
- 一次提交多个任务，然后我就直接执行下一行代码
- 喊朋友吃饭，朋友说知道了，就去忙他的事了，然后你就忙自己的事去了
#同步异步举例：
- 同步：先告诉一个老师完成写书的任务，我在原地等待，等他两个月后完成了，我再发布下一个写书的任务给下一个老师
- 异步：直接将三个任务告诉三个老师，我就忙我自己的了，直到三个老师完成任务之后告诉我
#同步意义
- 同步意味着顺序、统一的时间轴
- 是指完成事务的逻辑，先执行第一个事务，如果阻塞了，会一直等待，直到这个事务完成，再执行第二个事务，协同步调，按预定的先后次序进行运行
- 个任务的完成需要依赖另外一个任务时，只有等待被依赖的任务完成后，依赖的任务才能算完成，这是一种可靠的任务序列
#同步-多个线程有序执行
import threading
def run1():
while 1:
lock1.acquire()
print("111")
lock2.release()
def run2():
while 1:
lock2.acquire()
print("2222222")
lock3.release()
def run3():
while 1:
lock3.acquire()
print("33333333333")
lock1.release()
lock1 = threading.Lock()
lock2 = threading.Lock()
lock2.acquire() ##第一步：创建第二把锁就给锁上
lock3 = threading.Lock()
lock3.acquire() ##第二步：创建第三把锁就给锁上
t1 = threading.Thread(target=run1)
t2 = threading.Thread(target=run2)
t3 = threading.Thread(target=run3)
t1.start()
t2.start()
t3.start()
#生产者消费者模式
- 在线程世界⾥， ⽣产者就是⽣产数据的线程， 消费者就是消费数据的线程（做包子，吃包子），经常会出现生产数据的速度大于消费数据的速度，或者生产速度跟不上消费速度
- ⽣产者消费者模式是通过⼀个容器（缓冲区）来解决⽣产者和消费者的强耦合问题
- 例如两个线程共同操作一个列表，一个放数据，一个取数据
- ⽣产者和消费者彼此之间不直接通讯， ⽽通过阻塞队列来进⾏通讯
#同步-消息队列
-
class queue.Queue(maxsieze = 0)
- FIFO(先进先出)队列的构造器。maxsize为一个整数,表示队列的最大条目数，可用来限制内存的使用。
- 一旦队列满，插入将被阻塞直到队列中存在空闲空间。如果maxsize小于等于0，队列大小为无限。maxsize默认为0
from queue import Queue import threading def w1(): while 1: if q1.qsize() <= 1000: for i in range(1,50): q1.put(f"后来的数据{i}") def r1(): while 2: if q1.qsize() >= 100: for i in range(60): print("拿出",q1.get()) q1 = Queue() #创建一个队列，线程中能用，进程中不能用 for i in range(1,500): q1.put(f"初始数据{i}") for i in range(500): print(q1.get()) t1 = threading.Thread(target=r1) t2 = threading.Thread(target=w1) t1.start() t2.start()
#异步意义
- 异步则意味着乱序、效率优先的时间轴
- 处理调用这个事务的之后，不会等待这个事务的处理结果，直接处理第二个事务去了，通过状态、回调来通知调用者处理结果
- 对于I/O相关的程序来说，异步编程可以大幅度的提高系统的吞吐量，因为在某个I/O操作的读写过程中，系统可以先去处理其它的操作（通常是其它的I/O操作）
- 不确定执行顺序
#异步1-无需等待线程执行
import threading,time
def run(num):
print(f"线程{num}开始执行")
time.sleep(3)
print(f"线程{num}开始结束")
t1 = threading.Thread(target=run,args=(1,))
t2 = threading.Thread(target=run,args=(2,))
t1.start()
t2.start()
异步2-通过循环控制
import threading,time
num = 0
def run1():
global num
while 1:
if lock.acquire(False):
for i in range(50):
num += 1
print("1",num)
lock.release()
break
else:
print("111赶紧滚吧")
def run2():
global num
while 1:
if lock.acquire(False):
for i in range(30):
num += 1
print("2",num)
lock.release()
break
else:
print("222赶紧滚吧")
lock = threading.Lock()
t1 = threading.Thread(target=run1)
t2 = threading.Thread(target=run2)
t1.start()
t2.start()
异步3-回调机制
from multiprocessing import Pool
import random
import time
def run(f):
for i in range(1,5):
print(f"{f}下载文件{i}")
time.sleep(3)
return "下载完成"
def show(msg):
print(msg)
if __name__ == "__main__":
p = Pool(3) #指定进程池最高进程数
p.apply_async(func=run,args=("线程1",),callback=show)
##当func执行完毕后，return的东西会给到回调函数callback
p.apply_async(func=run,args=("线程2",),callback=show)
p.close()
p.join()
>>>
线程1下载文件1
线程2下载文件1
线程1下载文件2
线程2下载文件2
线程1下载文件3
线程2下载文件3
线程1下载文件4
线程2下载文件4
下载完成
下载完成
#协程
- 程序员YY出来的比线程更小的执行单元(微线程)，操作系统层面没有协程概念
- 微线程，用户级别的，用户程序负责任务之间的IO操作
- 一个线程作为容器里面可放置多个协程
- 只切换函数调用即可完成多线程，可以减少cpu的切换
- 协程自己主动让出CPU
- 需要安装：pip3 install greenlet
#greenlet
from greenlet import greenlet
import time
def run1():
while 1:
num = 0
for i in range(1,101):
num += 1
print("run1",num)
t2.switch()
a = input("run1请输入：")
print(a)
time.sleep(2)
def run2():
while 1:
num = 0
for i in range(1,101):
num += 1
print("run2",num)
t1.switch() ##调到run1执行的位置继续执行
a = input("run2请输入")
print(a)
time.sleep(2)
t1 = greenlet(run1) ##创建一个greenlet对象
t2 = greenlet(run2)
t1.switch() #执行run1函数
#gevent
- gevent比greenlet更强大
- 原理就是gevent遇到IO操作就会切换到其他的greenlet，等IO操作完成后，再在适当的时候切换回来继续执行
- 进程线程的任务切换是由操作系统自行切换的，人无法干预
- 协程可以通过自己的程序来进行切换，自己能够控制
- gevent只有遇到模块能够识别的IO操作的时候，程序才会进行任务切换，实现并发效果，如果所有程序都没IO操作，那么就基本属于串行了
import gevent
def run1():
while 1:
print("______run1______")
gevent.sleep(3) #模拟耗时操作#当协程遇到耗时操作会自动将控制权交给其他协程
def run2():
while 1:
print("______run2______")
gevent.sleep(3) #每当遇到耗时操作，会自动转到其他协程
g1 = gevent.spawn(run1) #创建一个gevent对象(创建一个协程) 创建的同时就开始执行了
g2 = gevent.spawn(run2)
g1.join() #等待线程执行结束
g2.join()