并发编程介绍
串行、并行与并发的区别
进程、线程、协程的区别
1. 进程 (Process)
- 定义：进程是操作系统为运行中的程序分配的基本单位。每个进程都有独立的地址空间和资源（如内存、文件句柄等）。
- 特点：
- 进程是资源分配的基本单位，具有独立内存空间。
- 进程之间的通信（IPC）相对复杂，通常需要使用管道、套接字等机制。
- 进程的创建和销毁开销较大。
2. 线程 (Thread)
- 定义：线程是进程中的一个执行单元，多个线程共享同一进程的内存和资源。
- 特点：
- 线程是程序执行的最小单位，一个进程可以拥有多个线程，这些线程共享同一进程的地址空间。
- 线程之间的通信相对简单，可以直接访问共享数据。
- 线程的创建和切换开销比进程小，但线程间的竞争和同步问题需要处理。
3. 协程 (Coroutine)
- 定义：协程是一种轻量级的用户态线程，可以在单个线程内实现多个任务的并发。
- 特点：
- 协程并不像线程那样由操作系统调度，而是由程序员控制，通常通过特定的语言特性实现。
- 协程允许在执行时暂停和恢复，非常适合处理I/O密集型任务，能够提高程序的并发性能。
- 协程的调度开销更低，并可以通过异步编程简化回调地狱的问题。
同步和异步介绍
同步和异步强调的是消息通信机制
同步：A调用B，等待B返回结果后，A继续执行
异步：A调用B，A继续执行，不等待B返回结果；B有结果了，通知A，A再做处理
线程Thread
什么是线程
线程主要是共用堆区的资源，而每个线程有自己的栈区，进程通常拥有独立的堆区和栈区。
线程（Thread）特点：
- 线程（Thread）是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位
- 现成是程序执行的最小单位，而进程是操作系统分配资源的最小单位。
- 一个进程由一个或多个线程组成，线程是一个进程中代码的不同执行路线。
- 拥有自己独立的栈和共享的堆，共享堆，不共享栈（每个线程有一个独立的栈），标准线程由操作系统调度；
- 调度和切换：线程上下文切换比进程上下文切换要快得多。（每个进程都有一个独立的堆，所有进程都共享一个堆，所以线程的切换要比进程快得多。）
线程的创建方式
python的标准库提供了两个模块：thread和threading，thread是低级模块，threading是高级模块，对thread进行了封装。绝大多数情况下，我们只需要使用threading这个高级模块。
现成的创建可以通过分为两种方式：
- 方法包装
- 类包装
现成的统一执行通过start()方法
线程的创建方式(方法包装)
# -*- coding: utf-8 -*-
from threading import Thread
import time
def func1(name):
print(f"线程{name},start")
for i in range(3):
print(f"线程:{name},{i}")
time.sleep(1)
print(f"线程{name},end")
if __name__ == '__main__':
print("主线程，start")
# 创建线程
t1 = Thread(target=func1, args=("a1",))
t2 = Thread(target=func1, args=("a2",))
# 启动线程
t1.start()
t2.start()
print("主线程,end")
为什么在这里的时候，会发生线程a2，start主线程，end重合到了一起，是因为两个线程会去抢夺资源导致，在打印换行的时候，控制台资源被抢夺了，此时的线程都是独立的。
类包装创建线程
# -*- coding: utf-8 -*-
import time
from threading import Thread
class MyThread(Thread):
def __init__(self, name):
Thread.__init__(self)
self.name = name
def run(self): # 这是重写方法
print(f"线程{self.name},start") # 线程创建开始就会执行的语句
for i in range(3):
print(f"线程{self.name},{i}")
time.sleep(2)
print(f"线程{self.name},end")
if __name__ == '__main__':
print("主线程，start")
# 创建线程
t1 = MyThread("t1")
t2 = MyThread("t2")
# 启动线程
t1.start()
t2.start()
print("主线程,end")
什么是重写方法？在 Python 中，重写方法（或称为方法重写）是指在子类中重新定义父类中已经定义过的方法。
重写方法
class Animal:
def speak(self):
return "Animal speaks"
class Dog(Animal):
def speak(self):
return "Bark"
class Cat(Animal):
def speak(self):
return "Meow"
# 示例
animal = Animal()
dog = Dog()
cat = Cat()
print(animal.speak()) # 输出: Animal speaks
print(dog.speak()) # 输出: Bark
print(cat.speak()) # 输出: Meow join()
之前的代码，主线程不会等待子线程结束。如需等待子线程结束后，再结束主线程，可使用join()方法。
# -*- coding: utf-8 -*-
from threading import Thread
from time import sleep
def func1(name):
for i in range(3):
print(f"thread:{name}:{i}")
sleep(1)
if __name__ == '__main__':
print("主线程，start")
# 创建线程
t1 = Thread(target=func1, args=("t1",))
t2 = Thread(target=func1, args=('t2',))
# 启动线程
t1.start()
t2.start()
# 主线程会等待t1,t2结束后，再往下执行
t1.join()
t2.join()
print("主线程,end")
主线程会等待子线程运行结束过后，才会结束主线程。
守护线程
在行为上还有一种叫守护线程，主要的特征是它的生命周期。主线程死亡，它也会随之死亡。在python中，现成通过setDaemon(True|False)来设置是否守护线程。
守护线程的作用：
守护线程作用是为其他线程提供便利服务，守护线程最典型的应用就是GC（垃圾收集器）
# -*- coding: utf-8 -*-
from threading import Thread
from time import sleep
class MyThread(Thread):
def __init__(self, name):
Thread.__init__(self)
self.name = name
def run(self):
for i in range(3):
print(f"thread:{self.name}:{i}")
sleep(1)
if __name__ == '__main__':
# 创建线程(类的方式)
t1 = MyThread('t1')
# t1设置为守护线程
t1.daemon = True # t1.setDaemon(True)3.10后被废弃，可以直接使用这里的
# t1.setDaemon(True)
# 启动线程
t1.start()
print("主线程,end")
全局锁GIL问题
在python中，无论你有多少核，在Cpython解释器中永远都是假
象。无论你是4核，8核，还是16核.......不好意思，同一时间执行的
线程只有一个线程，它就是这个样子的。这个是python的一个开发
时候，设计的一个缺陷，所以说python中的线程是“含有水分的线
程”
Python GIL（Global Interpreter Lock）
Python代码的执行由Python 虚拟机(也叫解释器主循环，CPython
版本)来控制，Python 在设计之初就考虑到要在解释器的主循环
中，同时只有一个线程在执行，即在任意时刻，只有一个线程在解
释器中运行。对Python 虚拟机的访问由全局解释器锁（GIL）来控
制，正是这个锁能保证同一时刻只有一个线程在运行。
GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念,同样一段代码可以通过CPython，PyPy，Psyco等不同的Python执行环境来执行,就没有GIL的问题。然而因为CPython是大部分环境下默认的Python执行环境。所以在很多人的概念里CPython就是Python，也就想当然的把GIL归结为Python语言的缺陷
线程同步和互斥锁
在现实生活中，我们会遇到“同一个资源，多个人都想使用”的问题。 比如：教室里，只有一台电脑，多个人都想使用。天然的解决办法就是，在电脑旁边，大家排队。前一人使用完后，后一人再使用。再比如，上厕所排队。
线程同步的概念
处理多线程问题时，多个线程访问同一个对象，并且某些线程
还想修改这个对象。 这时候，我们就需要用到“线程同步”。 线
程同步其实就是一种等待机制，多个需要同时访问此对象的线
程进入这个对象的等待池形成队列，等待前面的线程使用完毕
后，下一个线程再使用。
【示例】多线程操作同一个对象(未使用线程同步)
# -*- coding: utf-8 -*-
from threading import Thread
from time import sleep
class Account(object):
def __init__(self, money, name):
self.money = money
self.name = name
class Drawing(Thread):
def __init__(self, drawdingNum, account):
Thread.__init__(self)
self.drawingNum = drawdingNum
self.account = account
self.expenseTotal = 0
def run(self):
if self.account.money - self.drawingNum < 0:
return
sleep(1)
self.account.money -= self.drawingNum
self.expenseTotal += self.drawingNum
print(f"账户:{self.account.name},余额是:{self.account.money}")
print(f"账户:{self.account.name},总共取了：{self.expenseTotal}")
if __name__ == '__main__':
a1 = Account(100, "laoyang")
draw1 = Drawing(80, a1)
draw2 = Drawing(80, a1)
draw1.start()
draw2.start()
没有线程同步机制，两个线程同时操作同一个账户对象，竟然从只有100元的账户，轻松取出80*2=160元，账户余额竟然成为了-60。这么大的问题，显然银行不会答应的。
我们可以通过“锁机制”来实现线程同步问题，锁机制有如下几个要点：
- 必须使用同一个锁对象
- 互斥锁的作用就是保证同一时刻只能有一个线程去操作共享数据，保证共享数据不会出现错误问题
- 使用互斥锁的好处确保某段关键代码只能由一个线程从头到尾完整地去执行
- 使用互斥锁会影响代码的执行效率
- 同时持有多把锁，容易出现死锁的情况
互斥锁是什么？
互斥锁: 对共享数据进行锁定，保证同一时刻只能有一个线程去操作。
注意: 互斥锁是多个线程一起去抢，抢到锁的线程先执行，没有抢到锁的线程需要等待，等互斥锁使用完释放后，其它等待的线程再去抢这个锁。
threading 模块中定义了 Lock 变量，这个变量本质上是一个函数，通过
调用这个函数可以获取一把互斥锁。
【示例】多线程操作同一个对象(增加互斥锁，使用线程同步)
# -*- coding: utf-8 -*-
from threading import Thread, Lock
from time import sleep
class Account(object):
def __init__(self, money, name):
self.money = money
self.name = name
class Drawing(Thread):
def __init__(self, drawingNum, account):
Thread.__init__(self)
self.drawingNum = drawingNum
self.account = account
self.expenseTotal = 0
def run(self):
lock1.acquire()
if self.account.money - self.drawingNum < 0:
return
sleep(1)
self.account.money -= self.drawingNum
self.expenseTotal += self.drawingNum
lock1.release()
print(f"账户：{self.account.name},余额是：{self.account.money}")
print(f"账户：{self.account.name},总共取了:{self.expenseTotal}")
if __name__ == '__main__':
a1 = Account(100, 'laoyang')
lock1 = Lock() # 创建一个锁对象
draw1 = Drawing(80, a1)
draw2 = Drawing(80, a1)
draw1.start()
draw2.start()
死锁
在多线程程序中，死锁问题很大一部分是由于一个线程同时获取多个锁造成的。举例：
有两个人都要做饭，都需要“锅”和“菜刀”才能炒菜。
# -*- coding: utf-8 -*-
from threading import Thread, Lock
from time import sleep
lock1 = Lock()
lock2 = Lock()
def fun1():
lock1.acquire()
print('fun1拿到了菜刀')
sleep(2)
lock2.acquire()
print('fun1拿到了锅')
lock2.release()
print('fun1释放了锅')
lock1.release()
print('func释放菜刀')
def fun2():
lock2.acquire()
print('fun2拿到了锅')
lock1.acquire()
print('fun2拿到了菜刀')
lock1.release()
print('fun2释放了菜刀')
lock2.release()
print('fun2释放了锅')
if __name__ == '__main__':
lock1 = Lock()
lock2 = Lock()
t1 = Thread(target=fun1)
t2 = Thread(target=fun2)
t1.start()
t2.start()
其实逻辑上应该是整个做菜的过程，做菜要同时拿到锅和菜才能炒菜，但是这里设计的是为了第一时间就能触发死锁的机制，所有这样写了代码。
死锁的解决方法
死锁是由于“同步块需要同时持有多个锁造成”的，要解决这个问题，思路很简单，就是：同一个代码块，不要同时持有两个对象锁。
信号量(Semaphore)
互斥锁使用后，一个资源同时只有一个线程访问。如果某个资源，我们同时想让N个(指定数值)线程访问？这时候，可以使用信号量。
信号量控制同时访问资源的数量。信号量和锁相似，锁同一时间只允许一个对象(进程)通过，信号量同一时间允许多个对象(进程)通过。
应用场景
- 在读写文件的时候，一般只能只有一个线程在写，而读可以有多个线程同时进行，如果需要限制同
时读文件的线程个数，这时候就可以用到信号量了（如果用互斥锁，就是限制同一时刻只能有一个
线程读取文件）。 - 在做爬虫抓取数据时。
底层原理
信号量底层就是一个内置的计数器。每当资源获取时(调用acquire)计数器-1,资源释放时(调用release)计数器+1。
# -*- coding: utf-8 -*-
from threading import Thread, Lock
from time import sleep
from multiprocessing import Semaphore
"""
一个房间一次只允许两个人通过
若不使用信号量，会造成所有人都进入这个房子
若只允许一人通过可以用锁-Lock()
"""
def home(name, se):
se.acquire() # 拿到了一把钥匙
print(f"{name}进入了房间")
sleep(3)
print(f'*****************{name}走出来房间')
se.release() # 还回了一把钥匙
if __name__ == '__main__':
se = Semaphore(2) # 创建信号量的对象，有两把钥匙
for i in range(7):
p = Thread(target=home, args=(f'tom{i}', se))
p.start()
其实这里的se=Semaphore(2)，这里传入的参数2，就想到于有一个signal=2，每次调用se.acquire就会让signal-1，默认在signal=0时，会导致se发生暂时的死循环，等到有线程归还了se.release,signal+1,这时候的signal=1，刚才执行死循环的线程就可以重新访问se.acquire。
事件(Event)
事件Event主要用于唤醒正在阻塞等待状态的线程;
原理
Event 对象包含一个可由线程设置的信号标志，它允许线程等待某些事件的发生。在初始情况下，event 对象中的信号标志被设置假。如果有线程等待一个 event 对象，而这个 event 对象的标志为假，那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个 event 对象的信号标志设置为真，它将唤醒所有等待个 event 对象的线程。如果一个线程等待一个已经被设置为真的 event 对象，那么它将忽略这个事件，继续执行
Event() 可以创建一个事件管理标志，该标志（event）默认为False，event对象主要有四种方法可以调用：
| 方法名 | 说明 |
| event.wait(timeout=None) | 调用该方法的线程会被阻塞，如果设置了timeout参数，超时后， 线程会停止阻塞继续执行； |
| event.set() | 将event的标志设置为True，调用wait方法的所有线程将被唤醒 |
| event.clear() | 将event的标志设置为False，调用wait方法的所有线程将被阻塞 |
| event.is_set() | 判断event的标志是否为True |
【示例】Event事件对象经典用法
# -*- coding: utf-8 -*-
# 小伙伴们，围着吃火锅，当菜上齐了，请客的主人说：开吃！
# 于是小伙伴一起动筷子，这种场景如何实现
import threading
from threading import Thread, Event
from time import sleep
event = Event()
def chihuoguo(name):
# 等待事件，进入等待阻塞状态
print(f'{name}已经启动')
print(f'小伙伴{name}已经进入了就餐状态！')
sleep(1)
event.wait()
# 收到事件后进入运行状态
print(f'{name}收到了通知了.')
print(f'小伙伴{name}开始吃咯！')
if __name__ == '__main__':
# 创建新线程
thread1 = threading.Thread(target=chihuoguo, args=("tom",))
thread2 = threading.Thread(target=chihuoguo, args=('cherry',))
# 开始线程
thread1.start()
thread2.start()
# 发送事件通知
print('--->>>主线程通知小伙伴开吃咯！')
sleep(2)
event.set()
生产者和消费者模式
多线程环境下，我们经常需要多个线程的并发和协作。这个时候，就需要了解一个重要的多线程并发协作模型“生产者/消费者模式”
什么是生产者？
生产者指的是负责生产数据的模块（这里模块可能是：方法、对象、线程、进程）。
什么是消费者？
消费者指的是负责处理数据的模块（这里模块可能是：方法、对象、线程、进程）
什么是缓冲区？
消费者不能直接使用生产者的数据，它们之间有个“缓冲区”。生产者将生产好的数据放入“缓冲区”，消费者从“缓冲区”拿要处理的数据。
缓冲区是实现并发的核心，缓冲区的设置有3个好处：
- 实现线程的并发协作
有了缓冲区以后，生产者线程只需要往缓冲区里面放置数据，而不需要管消费者消费的情况；同样，消费者只需要从缓冲区拿数据处理即可，也不需要管生产者生产的情况。 这样，就从逻辑上实现了“生产者线程”和“消费者线程”的分离。- 解耦了生产者和消费者
生产者不需要和消费者直接打交道- 解决忙闲不均，提高效率
生产者生产数据慢时，缓冲区仍有数据，不影响消费者消费；消费者处理数据慢时，生产者仍然可以继续往缓冲区里面放置数据
缓冲区和queue对象
从一个线程向另一个线程发送数据最安全的方式可能就是使用queue 库中的队列了。创建一个被多个线程共享的 Queue 对象，这些线程通过使用 put() 和 get() 操作来向队列中添加或者删除元素。
Queue 对象已经包含了必要的锁，所以你可以通过它在多个线程间多安全地共享数据。
【示例】生产者消费者模式典型代码
# -*- coding: utf-8 -*-
from queue import Queue
from threading import Thread
from time import sleep
queue = Queue()
def producer():
num = 1
while True:
if queue.qsize() < 5:
print(f'生产：{num}号，大馒头')
queue.put(f'大馒头:{num}号')
num += 1
else:
print('馒头框满了，等待人来消费')
sleep(1)
def consumer():
while True:
print(f'获取馒头：{queue.get()}')
sleep(1)
if __name__ == '__main__':
queue = Queue()
t = Thread(target=producer)
t.start()
c = Thread(target=consumer)
c.start()
c2 = Thread(target=consumer)
c2.start()
进程Process
什么是进程
进程(Process)：拥有自己独立的堆和栈，既不共享堆，也不共享栈，进程由操作系统调度；进程切换需要的资源很最大，效率低。
对于操作系统来说，一个任务就是一个进程（Process），比如打开一个浏览器就是启动一个浏览器进程，就启动了一个记事本进程，打开两个记事本就启动了两个记事本进程，打开一个Word就启动了一个Word进程。
进程的优缺点
进程的优点：
- 可以使用计算机多核，进行任务的并行执行，提高执行效率
- 运行不受其他进程影响，创建方便
- 空间独立，数据安全
进程的缺点：
- 进程的创建和删除消耗的系统资源较多
进程的创建方式(方法模式)
Python的标准库提供了个模块： multiprocessing
进程的创建可以通过分为两种方式：
- 方法包装
- 类包装
创建进程后，使用start()启动进程
【示例】方法模式创建进程
# -*- coding: utf-8 -*-
from multiprocessing import Process
import os
from time import sleep
def func1(name):
print("当前进程ID：", os.getpid())
print("父进程ID：", os.getpid())
print(f"process:{name} start")
sleep(3)
print(f"process:{name} end")
"""这是一个关于windows上多进程实现的bug。
在windows上，子进程会自动import启动它的这个文件，而
在import的时候是会自动执行这些语句的。
如果不加__main__限制的话，就会无限递归创建子进程，进
而报错。
于是import的时候使用 __name__ =="__main__" 保护
起来就可以了"""
if __name__ == '__main__':
print("当前进程ID：", os.getpid())
# 创建进程
p1 = Process(target=func1, args=('p1',))
p2 = Process(target=func1, args=('p2',))
p1.start()
p2.start()
进程的创建方式(继承Process类)
和使用Thread 类创建子线程的方式非常类似，使用 Process 类创建实例化对象，其本质是调用该类的构造方法创建新进程。Process类的构造方法格式如下
def __init__(self,group=None,target=None,name=None,args=(),kwargs={})
其中，各个参数的含义为：
- group ：该参数未进行实现，不需要传参；
- target ：为新建进程指定执行任务，也就是指定一个函数；
- name ：为新建进程设置名称；
- args ：为 target 参数指定的参数传递非关键字参数；
- kwargs ：为 target 参数指定的参数传递关键字参数。
【示例】类的方式创建进程
# -*- coding: utf-8 -*-
from multiprocessing import Process
from time import sleep
class MyProcess(Process):
def __init__(self, name):
Process.__init__(self) # 第二行代码 Process.__init__(self) 的作用是调用父类 Process 的构造函数（__init__ 方法）。
self.name = name
def run(self):
print(f"Process:{self.name} start")
sleep(3)
print(f"Process:{self.name} end")
if __name__ == '__main__':
# 创建进程
p1 = MyProcess("p1")
p2 = MyProcess("p2")
p1.start()
p2.start()
Queue实现进程间通信
前面讲解了使用 Queue 模块中的 Queue 类实现线程间通信，但要实现进程间通信，需要使用 multiprocessing 模块中的 Queue 类。
简单的理解 Queue 实现进程间通信的方式，就是使用了操作系统给开辟的一个队列空间，各个进程可以把数据放到该队列中，当然也可以从队列中把自己需要的信息取走。
【示例】使用Queue实现进程间通信的经典代码
# -*- coding: utf-8 -*-
from multiprocessing import Process, Queue
class MyProcess(Process):
def __init__(self, name, mq):
Process.__init__(self)
self.name = name
self.mq = mq
def run(self):
print("Process:{} start".format(self.name))
print('-----------', self.mq.get(), '--------')
self.mq.put(self.name)
print("Process:{} end".format(self.name))
if __name__ == '__main__':
# 创建进程列表
t_list = []
mq = Queue()
mq.put('1')
mq.put('2')
mq.put('3')
# 循环创建进程
for i in range(3):
t = MyProcess('p{}'.format(i), mq)
t.start()
t_list.append(t)
# 等待进程结束
for t in t_list:
t.join()
print(mq.get())
print(mq.get())
print(mq.get())
Pipe实现进程间通信
Pipe 直译过来的意思是“管”或“管道”，和实际生活中的管（管道）是非常类似的。
Pipe方法返回（conn1， conn2）代表一个管道的两个端。
Pipe方法有duplex参数，如果duplex参数为True（默认值），那么这个参数是全双工模式，也就是说conn1和conn2均可收发。若duplex为False，conn1只负责接收消息，conn2只负责
发送消息。send和recv方法分别是发送和接受消息的方法。例如，在全双工模式下，可以调用conn1.send发送消息，conn1.recv接收消息。如果没有消息可接收，recv方法会一直
阻塞。如果管道已经被关闭，那么recv方法会抛出EOFError。
【示例】使用Pipe管道实现进程间通信
# -*- coding: utf-8 -*-
import multiprocessing
from time import sleep
def func1(conn1):
sub_info = "Hello!"
print(f"进程1--{multiprocessing.current_process().pid}发送数据：{sub_info}")
sleep(1)
conn1.send(sub_info)
print(f"来自进程2:{conn1.recv()}")
sleep(1)
def func2(conn2):
sub_info = "你好！"
print(f"进程2--{multiprocessing.current_process().pid}发送数据:{sub_info}")
sleep(1)
conn2.send(sub_info)
print(f"来自进程1:{conn2.recv()}")
sleep(1)
if __name__ == '__main__':
# 创建管道
conn1, conn2 = multiprocessing.Pipe()
# 创建子进程
process1 = multiprocessing.Process(target=func1, args=(conn1,))
process2 = multiprocessing.Process(target=func2, args=(conn2,))
# 启动子进程
process1.start()
process2.start()
Manager管理器
管理器提供了一种创建共享数据的方法，从而可以在不同进程中共享
【示例】管理器Manager实现进程通信
# -*- coding: utf-8 -*-
from multiprocessing import Process, current_process
from multiprocessing import Manager
def func(name, m_list, m_dict):
m_dict['name'] = '老杨'
m_list.append('你好')
if __name__ == '__main__':
with Manager() as mgr: # with语句主要用于简化资源管理，确保在使用完资源后能够被正确地释放。这通常用于文件操作、数据库连接、网络连接等场景。
m_list = mgr.list()
m_dict = mgr.dict()
m_list.append('Hello!!')
# 两个进程不能直接相互使用对象，需要互相传递
p1 = Process(target=func, args=('p1', m_list, m_dict))
p2 = Process(target=func, args=('p2', m_list, m_dict))
p1.start()
p2.start()
p1.join() # 等待p1进程结束，主进程继续执行
print(m_list)
print(m_dict)
进程池(Pool)
Python提供了更好的管理多个进程的方式，就是使用进程池。
进程池可以提供指定数量的进程给用户使用，即当有新的请求提交到进程池中时，如果池未满，则会创建一个新的进程用来执行该请求；反之，如果池中的进程数已经达到规定最大值，那么该请求就会等待，只要池中有进程空闲下来，该请求就能得到执行。
使用进程池的优点：
- 提高效率，节省开辟进程和开辟内存空间的时间及销毁进程的时间
- 节省内存空间
| 类/方法 | 功能 | 参数 |
| Pool(processes) | 创建进程池 对象 |
processes表示进程池中有多少进程 |
| pool.apply_async(func,args,kwds) | 异步执行 ；将事件放 入到进程池 队列 |
func 事件函数 args 以元组形式给func传参 kwds 以字典形式给func传参 返回值：返回 一个代表进程池事件的对象，通过返回值的 get方法可以得到事件函数的返回值 |
| pool.apply(func,args,kwds) | 同步执行； 将事件放入 到进程池队 列 |
func 事件函数 args 以元组形式给func传参 kwds 以字典形式给func传参 |
| pool.close() | 关闭进程池 | |
| pool.join() | 回收进程池 | |
| pool.map(func,iter) | 类似于 python的 map函数， 将要做的事 件放入进程 池 |
func 要执行的函数 iter 迭代对象 |
【示例】进程池使用案例
# -*- coding: utf-8 -*-
from multiprocessing import Pool
import os
from time import sleep
def func1(name):
print(f"当前进程的ID:{os.getpid()},name:{name}")
sleep(2)
return name
def func2(args):
print(args)
if __name__ == '__main__':
pool = Pool(5)
pool.apply_async(func=func1, args=('sxt1',), callback=func2)
pool.apply_async(func=func1, args=('sxt2',), callback=func2)
pool.apply_async(func=func1, args=('sxt3',), callback=func2)
pool.apply_async(func=func1, args=('sxt4',))
pool.apply_async(func=func1, args=('sxt5',))
pool.apply_async(func=func1, args=('sxt6',))
pool.apply_async(func=func1, args=('sxt7',))
pool.apply_async(func=func1, args=('sxt8',))
pool.close()
pool.join()
【示例】使用with管理进程池
# -*- coding: utf-8 -*-
from multiprocessing import Pool
import os
from time import sleep
def func1(name):
print(f"当前进程的ID:{os.getpid()},{name}")
sleep(2)
return name
if __name__ == '__main__':
with Pool(5) as pool:
args = pool.map(func1, ('sxt1,', 'sxt2,', 'sxt3,', 'sxt4', 'sxt5,', 'sxt6,', 'sxt7,', 'sxt8,'))
for a in args:
print(a)
协程Coroutines
协程是什么
协程，Coroutines，也叫作纤程(Fiber)
协程，全称是“协同程序”，用来实现任务协作。是一种在线程中，比线程更加轻量级的存在，由程序员自己写程序来管理。
当出现IO阻塞时，CPU一直等待IO返回，处于空转状态。这时候用协程，可以执行其他任务。当IO返回结果后，再回来处理数据。充分利用了IO等待的时间，提高了效率。
一个故事说明进程、线程、协程的关系
乔布斯想开工厂生产手机，费劲力气，制作一条生产线，这个生产线上有很多的器件以及材料。一条生产线就是一个进程。只有生产线是不够的，所以找五个工人来进行生产，这个工人能够利用这些材料最终一步步的将手机做出来，这五个工人就是五个线程。
为了提高生产率，想到3种办法：
- 一条生产线上多招些工人，一起来做手机，这样效果是成倍增长，即单进程多线程方式
- 多条生产线，每个生产线上多个工人，即多进程多线程
- 乔布斯深入一线发现工人不是那么忙，有很多等待时间。于是规定：如果某个员工在等待生产线某个零件生产时，不要闲着，干点其他工作。也就是说：如果一个线程等待某些条件，可以充分利用这个时间去做其它事情，这就是：协程方式。
协程的核心(控制流的让出和恢复)
- 每个协程有自己的执行栈，可以保存自己的执行现场
- 可以由用户程序按需创建协程（比如：遇到io操作）
- 协程“主动让出（yield）”执行权时候,会保存执行现场(保存中断时的寄存器上下文和栈)，然后切换到其他协程
- 协程恢复执行（resume）时，根据之前保存的执行现场恢复到中断前的状态，继续执行，这样就通过协程实现了轻量的由用户态调度的多任务模型
协程和多线程比较
比如，有3个任务需要完成，每个任务都在等待I/O操作时阻塞自身。阻塞在I/O操作上所花费的时间已经用灰色框标示出来了。
- 在单线程同步模型中，任务按照顺序执行。如果某个任务因为I/O而阻塞，其他所有的任务都必须等待，直到它完成之后它们才能依次执行。
- 多线程版本中，这3个任务分别在独立的线程中执行。这些线程由操作系统来管理，在多处理器系统上可以并行处理，或者在单处理器系统上交错执行。这使得当某个线程阻塞在某个资源的同时其他线程得以继续执行。
- 协程版本的程序中，3个任务交错执行，但仍然在一个单独的线程控制中。当处理I/O或者其他昂贵的操作时，注册一个回调到事件循环中，然后当I/O操作完成时继续执行。回调描述了该如何处理某个事件。事件循环轮询所有的事件，当事件到来时将它们分配给等待处理事件的回调函数。
协程的优点
- 由于自身带有上下文和栈，无需线程上下文切换的开销,属于程序级别的切换，操作系统完全感知不到，因而更加轻量级；
- 无需原子操作的锁定及同步的开销；
- 方便切换控制流，简化编程模型
- 单线程内就可以实现并发的效果，最大限度地利用cpu，且可扩展性高，成本低（注：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理）
asyncio协程是写爬虫比较好的方式。比多线程和多进程都好.开辟新的线程和进程是非常耗时的。
协程的缺点
- 无法利用多核资源：协程的本质是个单线程,它不能同时将 单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上。
- 当然我们日常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。
【示例】不使用协程执行多个任务
# -*- coding: utf-8 -*-
import time
def func1():
for i in range(3):
print(f"北京:第{i}次打印啦")
time.sleep(1)
return "func1执行完毕"
def func2():
for k in range(3):
print(f"上海:第{k}打印了")
time.sleep(1)
return "func2执行完毕"
def main():
func1()
func2()
if __name__ == '__main__':
start_time = time.time()
main()
end_time = time.time()
print(f"耗时{end_time - start_time}") # 不使用协程
asyncio实现协程(重点)
- 正常的函数执行时时不会中断的，所以那你要写一个能够中断的函数，就需要加asyncio
- async 用来声明一个函数为异步函数，异步函数的特点是能在函数执行过程中挂起，去执行其他异步函数，等到挂起条件（假设挂起条件是 sleep(5) ）消失后，也就是5秒到了再回来执行
- await 用来用来声明程序挂起，比如异步程序执行到某一步时需要等待的时间很长，就将此挂
起，去执行其他的异步程序。 - asyncio 是python3.5之后的协程模块，是python实现并发重要的包，这个包使用事件循环驱动实现并发。
【示例】asyncio异步IO的典型使用方式
# -*- coding: utf-8 -*-
import asyncio
import time
async def func1():
for i in range(3):
print(f"北京：第{i}次打印啦")
await asyncio.sleep(1)
return "func1执行完毕"
async def func2():
for k in range(3):
print(f"上海:第{k}次打印了")
await asyncio.sleep(1)
return "func2执行完毕"
async def main():
res = await asyncio.gather(func1(), func2())
print(res)
if __name__ == '__main__':
start_time = time.time()
asyncio.run(main())
end_time=time.time()
print(f"耗时{end_time-start_time}")
这个asyncio这个库会将申明的异步函数中return的值以列表的形式返回。