1.协程

协程，又称微线程。协程是python个中另外一种实现多任务的方式，只不过比线程更小占用更小执行单元（理解为需要的资源）。为啥说它是一个执行单元，因为它自带CPU上下文。这样只要在合适的时机，我们可以把一个协程切换到另一个协程。只要这个过程中保存或恢复 CPU上下文那么程序还是可以运行的。

通俗的理解：在一个线程中的某个函数，可以在任何地方保存当前函数的一些临时变量等信息，然后切换到另外一个函数中执行，注意不是通过调用函数的方式做到的，并且切换的次数以及什么时候再切换到原来的函数都由开发者自己确定。

协程,也称为用户级线程，在不开辟线程的基础上完成多任务，也就是在单线程的情况下完成多任务，多个任务按照一定顺序交替执行通俗理解只要在def里面只看到一个yield关键字表示就是协程。

在一个线程中如果遇到IO等待时间,线程不会傻傻等,利用空闲的时候再去干点其他事情;
在这里插入图片描述

（1）协程的发展历程

Python 对协程的支持经历了多个版本：

Python2.x 对协程的支持比较有限，通过 yield 关键字支持的生成器实现了一部分协程的功能但不完全。
第三方库 gevent 对协程有更好的支持。
Python3.4 中提供了 asyncio 模块。
Python3.5 中引入了 async/await 关键字。
Python3.6 中 asyncio 模块更加完善和稳定。
Python3.7 中内置了 async/await 关键字。

（2）协程的特点

单线程内切换，适用于IO密集型程序中，可以最大化IO多路复用的效果。
无法利用多核。
协程间完全同步，不会并行。不需要考虑数据安全。在单线程利用CPU和IO同时执行的原理，实现函数异步执行。

（3）协程是否需要加锁

https://zhuanlan.zhihu.com/p/169426477
同一个线程内的协程不需要加锁，不同线程内的协程就需要加锁。

2.协程和线程差异

在实现多任务时, 线程切换从系统层面远不止保存和恢复 CPU上下文这么简单。操作系统为了程序运行的高效性每个线程都有自己缓存Cache等等数据，操作系统还会帮你做这些数据的恢复操作。所以线程的切换非常耗性能。但是协程的切换只是单纯的操作CPU的上下文，所以一秒钟切换个上百万次系统都抗的住。

一个进程至少有一个线程，进程里面可以有多个线程
一个线程里面可以有多个协程

3. 协程知识点

https://zhuanlan.zhihu.com/p/169426477

（1）事件循环

理解成一个死循环,去检测并执行某些代码。

任务列表=[任务1,任务2,任务3 ..... ]
while True:
  可执行的人物列表,已完成的任务列表=去任务列表中检查所有任务,将'可执行'和'已完成'的任务返回
  for 就绪任务 in 可执行任务列表:
    执行已就绪的任务
  for 已完成的任务 in 已完成的任务列表
    任务列表中移除 已完成
如果 任务列表 中的任务都已完成, 终止循环

① 旧的做法

import asyncio

# 去一个获取事件循环
loop = asyncio.get_event_loop()
tasks = [asyncio.ensure_future(say_after('hello', i)) for i in range(4)]
# 将任务放到'任务列表'
loop.run_until_complete(asyncio.wait(tasks))

② 新的做法

# coding=utf-8
import asyncio
import time
async def say_after(what, delay):
    await asyncio.sleep(delay)
    print(what)
async def main():
    # create_task封装了事件循环
    tasks = []
    task1 = asyncio.create_task(say_after('hello', 1))
    task2 = asyncio.create_task(say_after('world', 2))
    tasks.append(task1)
    tasks.append(task2)
    print(f"程序于 {time.strftime('%X')} 开始执行")
    await asyncio.wait(tasks)
    print(f"程序于 {time.strftime('%X')} 执行结束")

asyncio.run(main())

（2）协程函数与协程对象

协程函数:定义函数时候 async def 函数名称为协程函数
协程对象:执行协程函数() 得到线程对象.

async def func(): # 协程方法
  pass

result=func() # 协程对象,内部代码不会执行的

如果想要运行携程函数内部代码,必须要将协程对象交给事件循环来处理。

import asyncio
async def func(): # 协程方法
  print('函数哈')

# 旧的写法-----------------------
# 获取事件循环
# loop=asyncio.get_event_loop()
# 协程对象,内部代码不会执行的
# tasks = [asyncio.ensure_future(func()) for i in range(4)]
# 将任务放到'任务列表'
# loop.run_until_complete(asyncio.wait(tasks))


# python3.7及以上可以使用最新写法：------------------
async def main():
    tasks = []
    task1 = asyncio.create_task(func()))
    tasks.append(task1)
    await asyncio.wait(tasks)


asyncio.run(main())

（3）await 【重要，新版本写法】

await+可等待的对象(协程对象,Future,Task对象) ~=（相当于） IO等待；await 就是等待对应的值得到结果后再继续向下走。

await 一个对象;
await asyncio.wait(多个可等待对象的列表);

import asyncio
async def others():
  print('start')
  await asyncio.sleep(2)
  print('end')
  return  '返回值'

async def func():
  print('执行协程函数内部代码')
  respo1=await others()
  print('IO请求结束结果为1',respo1)
  respo2=await others()
  print('IO请求结束结果为2',respo2)

asyncio.run(func())

执行协程函数内部代码
start
end
IO请求结束结果为1 返回值
start
end
IO请求结束结果为2 返回值

（4）Task对象【重要，新版本写法】

在事件循环中添加多个任务的。Task用于并发调度协程,通过asyncio.create_task(协程对象) 的方式创建Task对象,这样可以让协程加入事件循环中等待被调度执行.除了使用asyncio.create_task()函数以外,还可以用底层级的loop.create_task()或asyncio.ensure_future()函数.不建议手动实例化Task对象。

注意:asyncio.create_task()函数在python3.7中被加入,在python3.7之前可以改用底层级的asyncio.ensure_future()函数。

import asyncio

async def func():
  print(1)
  await asyncio.sleep(2)
  print(2)
  return 'func111'
async def main():
  print('start main')
  # 创建task对象,将当前执行func函数任务添加到事件循环中.
  task_list=[
    asyncio.create_task(func(),name='n1'),
    asyncio.create_task(func(),name='n2')
  ]
  print('main end')
  don,pending=await asyncio.wait(task_list,timeout=None) 
  # 最多等待1秒,如果任务没有完成,那么返回结果就是空的
  print(don)
  print(pending)
asyncio.run(main())

（5）asyncio.Future对象

Task继承Future,Task对象内部await结果的处理基于Future对象来的。Task和Future的区别:如果Future没有设置返回值那么await就会一直等待,Task函数执行完成后默认会调用set_result(‘666’)来结束。

下面是新旧写法的混合：

import asyncio
async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result('666')

async def main():
    #获取当前事件循环
    loop=asyncio.get_running_loop()
    # 创建一个任务(Future对象),这个任务什么也没有干
    fut=loop.create_future()
    await loop.create_task(set_after(fut))
    # 等待任务最终结果(Future对象),没有结果则会一直等待下去
    data=await asyncio.wait(fut)
    print(data)

asyncio.run(main())

（6）uvloop 事件循环【旧】

是asyncio的事件循环的代替方案，事件循环>默认asyncio的事件循环。

pip install uvloop

import uvloop
import asyncio
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 编写asyncio的代码 与之前一样
# 内部的事件循环自动化会变为uvloop
asyncio.run(...)

注意事项：一个asgi->uvicorn 内部使用的就是uvloop。

4.简述实现协程的几种方式

（1）协程 - yield关键字

import time

def work1():
    while True:
        print("---work1---")
        yield
        time.sleep(0.5)
        
        
def work2():
    while True:
        print("---work2---")
        yield
        time.sleep(0.5)
         
def main():
    w1 = work1()
    w2 = work2()
    while True:
        next(w1)
        next(w2)
        
if __name__ == '__main__':
    main()

（2）协程 - greenlet

为了更好使用协程来完成多任务，python中的greenlet模块对其封装，从而使得切换任务变的更加简单
安装：

pip install greenlet

import time
from greenlet import greenlet


def test1():
    while True:
        print("---A---")
        gr2.switch()  # 切换到gr2中运行
        time.sleep(0.5)
        
        
def test2():
    while True:
        print("---B---")
        gr1.switch()
        time.sleep(0.5)
         
gr1 = greenlet(test1)
gr2 = greenlet(test2)


gr1.switch()

（3）协程 - gevent

greenlet 已经实现了协程，但是这个还的人工切换，太麻烦了。python还有一个比greenlet更强大的并且能够自动切换任务的模块gevent。
其原理是当一个 greenlet 遇到IO(指的是input output 输入输出，比如网络、文件操作等)操作时，比如访问网络，就自动切换到其他的greenlet，等到IO操作完成，再在适当的时候切换回来继续执行。

由于IO操作非常耗时，经常使程序处于等待状态，有了gevent为我们自动切换协程，就保证总有greenlet在运行，而不是等待IO

安装方法：

pip install gevent

import gevent

def f(n):
    for i in range(n):
        print(gevent.getcurrent(),i)
        #用来模拟一个耗时操作，注意不是time模块中的sleep，注意和time.sleep(1)的结果区别
        gevent.sleep(1)

g1 = gevent.spawn(f,5)
g2 = gevent.spawn(f,5)
g3 = gevent.spawn(f,5)


g1.join()
g2.join()
g3.join()

或者给程序打补丁代替gevent.sleep(1)。

from gevent import monkey
import gevent
import random
import time

# 有耗时操作时需要
monkey.patch_all()  # 将程序中用到的耗时操作的代码，换为gevent中自己实现的模块

def coroutine_work(coroutine_name):
    for i in range(5):
        print(coroutine_name, i)
        time.sleep(random.random())

gevent.joinall([
        gevent.spawn(coroutine_work, "work1"),
        gevent.spawn(coroutine_work, "work2")
])

（4）协程 - asyncio装饰器【旧写法】

要实现异步并行，需要将协程函数打包成一个任务（Task）。注意:遇到IO等耗时操作会自动切换.

import asyncio
import time
@asyncio.coroutine
def func1():
    print(1)
    yield  from asyncio.sleep(3) # 遇到耗时后会自动切换到其他函数中执行
    print(2)

@asyncio.coroutine
def func2():
    print(3)
    yield  from asyncio.sleep(2)
    print(4)

@asyncio.coroutine
def func3():
    print(5)
    yield  from asyncio.sleep(2)
    print(6)

tasks=[
    asyncio.ensure_future( func1() ),
    asyncio.ensure_future( func2() ),
    asyncio.ensure_future( func3() )
]

# 协程函数使用 func1()这种方式是执行不了的
start=time.time()
loop=asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
# loop.run_until_complete(func1()) 执行一个函数
end=time.time()
print(end-start) # 只会等待3秒

（5）协程 - asyc&await 关键字

① 旧的写法

import asyncio
import time

async def func1():
    print(1)
    await asyncio.sleep(3) # 遇到耗时后会自动切换到其他函数中执行
    print(2)


async def func2():
    print(3)
    await asyncio.sleep(2)
    print(4)


async def func3():
    print(5)
    await asyncio.sleep(2)
    print(6)

tasks=[
    asyncio.ensure_future( func1() ),
    asyncio.ensure_future( func2() ),
    asyncio.ensure_future( func3() )
]

# 协程函数使用 func1()这种方式是执行不了的
start=time.time()
loop=asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
# loop.run_until_complete(func1()) 执行一个函数
end=time.time()
print(end-start) # 只会等待3秒

② 新的写法

import asyncio
import time

async def func1():
    print(1)
    await asyncio.sleep(3) # 遇到耗时后会自动切换到其他函数中执行
    print(2)


async def func2():
    print(3)
    await asyncio.sleep(2)
    print(4)


async def func3():
    print(5)
    await asyncio.sleep(2)
    print(6)


async def main():
    tasks = [
        asyncio.create_task(func1()),
        asyncio.create_task(func2()),
        asyncio.create_task(func3()),
    ]
    await asyncio.wait(tasks)

# 协程函数使用 func1()这种方式是执行不了的
start=time.time()
asyncio.run(main())
end=time.time()
print(end-start) # 只会等待3秒

（6）注意事项

其实 asyncio 和 asyc 几乎是互相关联的。

（7）最新的实现方式

在这里插入图片描述

5.进程、线程、协程对比

https://zhuanlan.zhihu.com/p/169426477

进程是资源分配的单位；
线程是操作系统调度的单位；
进程切换需要的资源很最大，效率很低；
线程切换需要的资源一般，效率一般（当然了在不考虑GIL的情况下）；
协程切换任务资源很小，效率高；
多进程、多线程根据cpu核数不一样可能是并行的，但是协程是在一个线程中所以是并发。

协程是一种比线程更加轻量级的存在，协程不是被操作系统内核所管理，而完全是由用户态程序所控制。 协程与线程以及进程的关系如下图所示。可见，协程自身无法利用多核，需要配合进程来使用才可以在多核平台上发挥作用。
在这里插入图片描述

协程之间的切换不需要涉及任何 System Call（系统调用）或任何阻塞调用。
协程只在一个线程中执行，切换由用户态控制，而线程的阻塞状态是由操作系统内核来完成的，因此协程相比线程节省线程创建和切换的开销。
协程中不存在同时写变量的冲突，因此，也就不需要用来守卫关键区块的同步性原语，比如：互斥锁、信号量等，并且不需要来自操作系统的支持。

（1）协程的真实本质

协程通过 “挂起点” 来主动 yield（让出）CPU，并保存自身的状态，等候恢复。例如：首先在 funcA 函数中执行，运行一段时间后调用协程，协程开始执行，直到第一个挂起点，此后就像普通函数一样返回 funcA 函数。 funcA 函数执行一些代码后再次调用该协程，注意，协程这时就和普通函数不一样了。协程并不是从第一条指令开始执行而是从上一次的挂起点开始执行，执行一段时间后遇到第二个挂起点，这时协程再次像普通函数一样返回 funcA 函数，funcA 函数执行一段时间后整个程序结束。
在这里插入图片描述

（2）抢占式调度：多线程、多进程

在 I/O 密集型场景中，抢占式调度的解决方案是 “异步 + 回调” 机制。
在这里插入图片描述
其存在的问题是，在某些场景中会使得整个程序的可读性非常差。以图片下载为例，图片服务中台提供了异步接口，发起者请求之后立即返回，图片服务此时给了发起者一个唯一标识 ID，等图片服务完成下载后把结果放到一个消息队列，此时需要发起者不断消费这个 MQ 才能拿到下载是否完成的结果。
在这里插入图片描述

可见，整体的逻辑被拆分为了好几个部分，各个子部分都会存在状态的迁移，日后必然是 BUG 的高发地。

（3）用户态协同式调度：协程

而随着网络技术的发展和高并发要求，协程所能够提供的用户态协同调度机制的优势，在网络操作、文件操作、数据库操作、消息队列操作等重 I/O 操作场景中逐渐被挖掘。

协程将 I/O 的处理权从内核态的操作系统交还给用户态的程序自身。用户态程序在执行 I/O 时，主动的通过 yield（让出）CPU 的执行权给其他协程，多个协程之间处于平等、对称、合作的关系。
在这里插入图片描述

（4）协程的运行原理

当程序运行时，操作系统会为每个程序分配一块同等大小的虚拟内存空间，并将程序的代码和所有静态数据加载到其中。然后，创建和初始化 Stack 存储，用于储存程序的局部变量，函数参数和返回地址；创建和初始化 Heap 内存；创建和初始化 I/O 相关的任务。当前期准备工作完成后，操作系统将 CPU 的控制权移交给新创建的进程，进程开始运行。

一个进程可以有一个或多个线程，同一进程中的多个线程将共享该进程中的全部系统资源，如：虚拟地址空间，文件描述符和信号处理等等。但同一进程中的多个线程有各自的调用栈和线程本地存储。

在这里插入图片描述
协程通过 “挂起点” 来主动 yield（让出）CPU，并保存自身的状态，等候恢复。例如：首先在 funcA 函数中执行，运行一段时间后调用协程，协程开始执行，直到第一个挂起点，此后就像普通函数一样返回 funcA 函数。 funcA 函数执行一些代码后再次调用该协程，注意，协程这时就和普通函数不一样了。协程并不是从第一条指令开始执行而是从上一次的挂起点开始执行，执行一段时间后遇到第二个挂起点，这时协程再次像普通函数一样返回 funcA 函数，funcA 函数执行一段时间后整个程序结束。
在这里插入图片描述
协程之所可以能够 “主动让出” 和 “被恢复”，是解析器在函数运行时堆栈中保存了其运行的 Context（上下文）。

6. 案例

（1）geven并发下载

能够看到是先发送的获取baidu的相关信息，然后依次是itcast、itheima，但是收到数据的先后顺序不一定与发送顺序相同，这也就体现出了异步，即不确定什么时候会收到数据，顺序不一定。

from gevent import monkey
import gevent
import urllib.request

# 有耗时操作时需要
monkey.patch_all()

def my_downLoad(url):
    print('GET: %s' % url)
    resp = urllib.request.urlopen(url)
    data = resp.read()
    print('%d bytes received from %s.' % (len(data), url))

gevent.joinall([
        gevent.spawn(my_downLoad, 'http://www.baidu.com/'),
        gevent.spawn(my_downLoad, 'http://www.itcast.cn/'),
        gevent.spawn(my_downLoad, 'http://www.itheima.com/'),
])

（2）geven实现多个视频下载

from gevent import monkey
import gevent
import urllib.request

#有IO才做时需要这一句
monkey.patch_all()

def my_downLoad(file_name, url):
    print('GET: %s' % url)
    resp = urllib.request.urlopen(url)
    data = resp.read()

    with open(file_name, "wb") as f:
        f.write(data)

    print('%d bytes received from %s.' % (len(data), url))

gevent.joinall([
        gevent.spawn(my_downLoad, "1.mp4", 'http://oo52bgdsl.bkt.clouddn.com/05day-08-%E3%80%90%E7%90%86%E8%A7%A3%E3%80%91%E5%87%BD%E6%95%B0%E4%BD%BF%E7%94%A8%E6%80%BB%E7%BB%93%EF%BC%88%E4%B8%80%EF%BC%89.mp4'),
        gevent.spawn(my_downLoad, "2.mp4", 'http://oo52bgdsl.bkt.clouddn.com/05day-03-%E3%80%90%E6%8E%8C%E6%8F%A1%E3%80%91%E6%97%A0%E5%8F%82%E6%95%B0%E6%97%A0%E8%BF%94%E5%9B%9E%E5%80%BC%E5%87%BD%E6%95%B0%E7%9A%84%E5%AE%9A%E4%B9%89%E3%80%81%E8%B0%83%E7%94%A8%28%E4%B8%8B%29.mp4'),
])

（3）asyncio协程对比

① 同步的情况下

import requests
import time
def downlod_image(url):
    print('开始下载')
    res=requests.get(url).content
    print('下载完成')
    # 图片保存到本地文件
    fileName=url.rsplit('-')[-1]
    with open(fileName,'wb') as wf:
        wf.write(res)

url_list=[
    'http://pic.netbian.com/uploads/allimg/220629/224839-1656514119b359.jpg',
    'http://pic.netbian.com/uploads/allimg/220420/114427-16504262671afd.jpg',
    'http://pic.netbian.com/uploads/allimg/220623/234915-16559993552953.jpg'
]

if __name__=='__main__':
    start=time.time()
    for item in url_list:
        downlod_image(item)
    end=time.time()
    print('本次耗时',end-start) # 本次耗时 1.5057339668273926

本次耗时 0.6898803710937

② 协程方式情况下【新写法】

import asyncio

import requests
import time
import aiohttp
async def downlod_image(session,url):
    print('开始下载')
    async with session.get(url,verify_ssl=False) as res:
        content=await res.content.read()
        print('下载完成')
        # 图片保存到本地文件
        fileName=url.rsplit('-')[-1]
        with open(fileName,'wb') as wf:
            wf.write(content)

url_list=[
    'http://pic.netbian.com/uploads/allimg/220629/224839-1656514119b359.jpg',
    'http://pic.netbian.com/uploads/allimg/220420/114427-16504262671afd.jpg',
    'http://pic.netbian.com/uploads/allimg/220623/234915-16559993552953.jpg'
]
async def main():
    async with aiohttp.ClientSession() as session:
        tasks=[asyncio.create_task(downlod_image(session,url)) for url in url_list]
        await asyncio.wait(tasks)
if __name__=='__main__':
    start=time.time()
    asyncio.run(main())
    end=time.time()
    print("本次耗时",end-start) # 本次耗时 0.3856923580169678

本次耗时 0.4509727954864502

（4）asyncio+不支持异步的模块

① 旧写法

# 案例:asyncio+不支持异步的模块
import asyncio
import requests
import time
import aiohttp
async def downlod_image(url):
    print('开始下载')
    loop=asyncio.get_event_loop()
    # 开启线程进行请求后转换成协程进行等待
    future=loop.run_in_executor(None,requests.get,url)
    res=await future
    print('下载完成')
    # 图片保存到本地文件
    fileName=url.rsplit('-')[-1]
    with open(fileName,'wb') as wf:
        wf.write(res.content)

url_list=[
    'http://pic.netbian.com/uploads/allimg/220629/224839-1656514119b359.jpg',
    'http://pic.netbian.com/uploads/allimg/220420/114427-16504262671afd.jpg',
    'http://pic.netbian.com/uploads/allimg/220623/234915-16559993552953.jpg'
]
def main():
    tasks=[downlod_image(url) for url in url_list]
    loop=asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

if __name__=='__main__':
    start=time.time()
    main()
    end=time.time()
    print("本次耗时",end-start) # 本次耗时 0.4465653896331787

本次耗时 1.613558292388916

② 新写法

# 案例:asyncio+不支持异步的模块
import asyncio
import requests
import time
import aiohttp
async def downlod_image(url):
    print('开始下载')
    loop=asyncio.get_event_loop()
    # 开启线程进行请求后转换成协程进行等待
    future=loop.run_in_executor(None,requests.get,url)
    res=await future
    print('下载完成')
    # 图片保存到本地文件
    fileName=url.rsplit('-')[-1]
    with open(fileName,'wb') as wf:
        wf.write(res.content)


async def main():
    url_list = [
        'http://pic.netbian.com/uploads/allimg/220629/224839-1656514119b359.jpg',
        'http://pic.netbian.com/uploads/allimg/220420/114427-16504262671afd.jpg',
        'http://pic.netbian.com/uploads/allimg/220623/234915-16559993552953.jpg'
    ]
    tasks=[downlod_image(url) for url in url_list]
    # tasks=[asyncio.create_task(downlod_image(url)) for url in url_list]
    await asyncio.wait(tasks)


if __name__=='__main__':
    start=time.time()
    asyncio.run(main())
    end=time.time()
    print("本次耗时",end-start) # 本次耗时 0.4465653896331787

本次耗时 0.7298567295074463

（5）协程下载图片

① 旧写法

import asyncio
from pathlib import Path
import logging
from urllib.request import urlopen, Request
import os
from time import time
import aiohttp
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
CODEFLEX_IMAGES_URLS = ['https://codeflex.co/wp-content/uploads/2021/01/pandas-dataframe-python-1024x512.png',
                        'https://codeflex.co/wp-content/uploads/2021/02/github-actions-deployment-to-eks-with-kustomize-1024x536.jpg',
                        'https://codeflex.co/wp-content/uploads/2021/02/boto3-s3-multipart-upload-1024x536.jpg',
                        'https://codeflex.co/wp-content/uploads/2018/02/kafka-cluster-architecture.jpg',
                        'https://codeflex.co/wp-content/uploads/2016/09/redis-cluster-topology.png']
async def download_image_async(session, dir, img_url):
    download_path = dir / os.path.basename(img_url)
    async with session.get(img_url) as response:
        with download_path.open('wb') as f:
            while True:
                # 在 async 函数中使用 await 关键字表示等待 task 执行完成，也就是等待 yeild 让出控制权。
                # 同时，asyncio 使用事件循环 event_loop 来实现整个过程。
                chunk = await response.content.read(512)
                if not chunk:
                    break
                f.write(chunk)
    logger.info('Downloaded: ' + img_url)
# 使用 async 关键字声明一个异步/协程函数。
# 调用该函数时，并不会立即运行，而是返回一个协程对象，后续在 event_loop 中执行。
async def main():
    images_dir = Path("codeflex_images")
    Path("codeflex_images").mkdir(parents=False, exist_ok=True)
    async with aiohttp.ClientSession() as session:
        tasks = [(download_image_async(session, images_dir, img_url)) for img_url in CODEFLEX_IMAGES_URLS]
        await asyncio.gather(*tasks, return_exceptions=True)
if __name__ == '__main__':
    start = time()
    # event_loop 事件循环充当管理者的角色，将控制权在几个协程函数之间切换。
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(main())
    logger.info('Download time: %s seconds', time() - start)

2022-11-24 10:04:54,108 - __main__ - INFO - Download time: 33.83906531333923 seconds

② 新写法

import asyncio
from pathlib import Path
import logging
from urllib.request import urlopen, Request
import os
from time import time
import aiohttp
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
CODEFLEX_IMAGES_URLS = ['https://codeflex.co/wp-content/uploads/2021/01/pandas-dataframe-python-1024x512.png',
                        'https://codeflex.co/wp-content/uploads/2021/02/github-actions-deployment-to-eks-with-kustomize-1024x536.jpg',
                        'https://codeflex.co/wp-content/uploads/2021/02/boto3-s3-multipart-upload-1024x536.jpg',
                        'https://codeflex.co/wp-content/uploads/2018/02/kafka-cluster-architecture.jpg',
                        'https://codeflex.co/wp-content/uploads/2016/09/redis-cluster-topology.png']
async def download_image_async(session, dir, img_url):
    download_path = dir / os.path.basename(img_url)
    async with session.get(img_url) as response:
        with download_path.open('wb') as f:
            while True:
                # 在 async 函数中使用 await 关键字表示等待 task 执行完成，也就是等待 yeild 让出控制权。
                # 同时，asyncio 使用事件循环 event_loop 来实现整个过程。
                chunk = await response.content.read(512)
                if not chunk:
                    break
                f.write(chunk)
    logger.info('Downloaded: ' + img_url)
# 使用 async 关键字声明一个异步/协程函数。
# 调用该函数时，并不会立即运行，而是返回一个协程对象，后续在 event_loop 中执行。
async def main():
    images_dir = Path("codeflex_images")
    Path("codeflex_images").mkdir(parents=False, exist_ok=True)
    async with aiohttp.ClientSession() as session:
        tasks = [(download_image_async(session, images_dir, img_url)) for img_url in CODEFLEX_IMAGES_URLS]
        # tasks = [asyncio.create_task((download_image_async(session, images_dir, img_url))) for img_url in CODEFLEX_IMAGES_URLS]
        await asyncio.gather(*tasks, return_exceptions=True)
if __name__ == '__main__':
    start = time()
    # event_loop 事件循环充当管理者的角色，将控制权在几个协程函数之间切换。
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())
    logger.info('Download time: %s seconds', time() - start)

7. 异步操作数据

（1）异步redis

在使用python代码操作redis时，链接/操作/断开都是网络IO。

pip install aioredis

import aioredis
import asyncio
async def execute(address,password):
    print('start-',address)
    # 网络IO操作：先去链接47.93.4.189：6666，遇到IO自动切换任务，去连接47.93.4.198.2712
    redis=await aioredis.create_redis_pool(address,password=password)
    # 网络IO操作：遇到IO会自动切换任务
    await redis.hmset_dict('car',key=1,key2=2,key3=3)
    # 网络IO操作：遇到IO会自动切换任务
    result=await redis.hgetall('car',encoding='utf-8')
    print(result)
    redis.close()
    # 网络IO操作：遇到IO会自动切换任务
    await redis.wait_closed()
    print('end-',address)
task_list=[
    execute('redis://47.93.4.189：6666','123456'),
    execute('redis://47.93.4.198.2712','123456')
]

asyncio.run(asyncio.wait(task_list))

（2）异步MySQL

pip install aiomysql

import asyncio
import aiomysql
async def execute():
    # 网络IO操作：链接mysql
    conn=await aiomysql.connect(host='127.0.0.1',port=3306,user='root',password='123',db='mysql')
    # 网络IO操作，创建CURSOR
    cur=await conn.cursor()
    # 网络IO操作，执行SQL
    await cur.execute('select * from user')
    # 网络IO操作，执行SQL
    result=await cur.fetchall()
    print(result)
    # 网络IO操作，执行SQL
    await cur.close()
    conn.close()

asyncio.run(execute())

import asyncio
import aiomysql
async def execute(host,password):
    # 网络IO操作：链接mysql
    conn=await aiomysql.connect(host=host,port=3306,user='root',password=password,db='mysql')
    # 网络IO操作，创建CURSOR
    cur=await conn.cursor()
    # 网络IO操作，执行SQL
    await cur.execute('select * from user')
    # 网络IO操作，执行SQL
    result=await cur.fetchall()
    print(result)
    # 网络IO操作，执行SQL
    await cur.close()
    conn.close()
task_list=[
    execute('127.0.0.1','123456'),
    execute('192.168.3.112','1a23456')
]

asyncio.run(asyncio.wait(task_list))