标签：jpg   set   UNC   als   close   item   通过   状态   tap   

异步 asyncio、async、await

• 异步非阻塞

• tronado、fastapi、django3.x asgi、sanic、aiohttp都在异步——>提升性能

• 协程

• asyncio模块进行异步编程

• 实战案例

1.协程

协程不是计算机提供的，是程序员人为创造的。

协程（Coroutine），也可以被称为微线程，是一种用户态内的上下文切换技术。简而言之，其实就是通过一个线程实现代码块互相切换执行。例如：

def func1():
    print(1)
    ...
    print(2)

def func2():
    print(3)
    ...
    print(4)
    
func1()
func2()

实现协程有这么几种方法：

• greenlet，早期模块。
• yield关键字。
• asyncio装饰器(py3.4)
• async、await关键字(py3.5)【推荐】

1.1 greenlet实现协程

pip install greenlet

from greenlet import greenlet


def func1():
    print(1)      # 第2步：输出 1
    gr2.switch()  # 第3步：切换到 func2 函数
    print(2)      # 第6步：输出 2
    gr2.switch()  # 第7步：切换到 func2 函数，从上一次执行的位置继续向后执行


def func2():
    print(3)      # 第4步：输出 3
    gr1.switch()  # 第5步：切换到 func1 函数，从上一次执行的位置继续向后执行
    print(4)      # 第8步：输出 4


gr1 = greenlet(func1)
gr2 = greenlet(func2)
gr1.switch()  # 第一步：去执行 func1 函数

1.2 yield关键字(很少用)

def func1():
    yield 1
    yield from func2()
    yield 2


def func2():
    yield 3
    yield 4


f1 = func1()
for item in f1:
    print(item)

1.3 asyncio

import time
import asyncio


@asyncio.coroutine
def func1():
    print(1)
    yield from asyncio.sleep(2)  # 遇到IO耗时操作，自动化切换到tasks中的其他任务
    print(2)


@asyncio.coroutine
def func2():
    print(3)
    yield from asyncio.sleep(2)  # 遇到IO耗时操作，自动化切换到tasks中的其他任务
    print(4)


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

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
print(time.time()-t1)
>>>>>>>>
2.0024895668029785

注意：遇到IO阻塞自动切换

1.4 async & await关键字

import asyncio
import time

async def func1():
    print(1)
    # 网络IO请求，下载一张图片，阻塞中
    await asyncio.sleep(2)  # 遇到IO耗时操作，自动化切换到tasks中的其他任务
    print(2)



async def func2():
    print(3)
    # 网络IO请求，下载一张图片，阻塞中
    await asyncio.sleep(2)  # 遇到IO耗时操作，自动化切换到tasks中的其他任务
    print(4)


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

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
print(time.time()-t1)
>>>>>>>
2.0024681091308594

2.协程意义

在一个线程中如果遇到IO等待时间，线程不会傻傻等，利用等待时间再去干点其他的事。

案例：去下载三张图片（网络IO）

• 普通方式（同步方式）

  import requests
  
  def down_img(url):
      print("开始下载：",url)
      res = requests.get(url).content
      print("下载完成")
      file_name = url.split(‘,‘)[-1]
      with open(file_name, ‘wb‘) as f:
          f.write(res)
  
  
  if __name__ == ‘__main__‘:
      urls = [
          "https://dss2.bdstatic.com/70cFvnSh_Q1YnxGkpoWK1HF6hhy/it/u=3228549874,2173006364&fm=26&gp=0.jpg",
          "https://dss0.bdstatic.com/70cFuHSh_Q1YnxGkpoWK1HF6hhy/it/u=1280325423,1024589167&fm=26&gp=0.jpg",
          "https://dss1.bdstatic.com/70cFuXSh_Q1YnxGkpoWK1HF6hhy/it/u=3206689113,2237998950&fm=26&gp=0.jpg"
      ]
      for url in urls:
          down_img(url)   
  >>>>>>>>
  0.1765291690826416
  

• 协程方式（异步方式）

  import asyncio
  import aiohttp
  import time
  
  async def get_img(session, url):
      print("方式请求", url)
      async with session.get(url, verify_ssl=False) as response:
          res = await response.content.read()
          file_name = url.split(‘,‘)[-1]
          with open(file_name, ‘wb‘) as f:
              f.write(res)
      print("下载完成")
  
  
  async def main():
      async with aiohttp.ClientSession() as session:
          urls = [
              "https://dss2.bdstatic.com/70cFvnSh_Q1YnxGkpoWK1HF6hhy/it/u=3228549874,2173006364&fm=26&gp=0.jpg",
              "https://dss0.bdstatic.com/70cFuHSh_Q1YnxGkpoWK1HF6hhy/it/u=1280325423,1024589167&fm=26&gp=0.jpg",
              "https://dss1.bdstatic.com/70cFuXSh_Q1YnxGkpoWK1HF6hhy/it/u=3206689113,2237998950&fm=26&gp=0.jpg"
          ]
          tasks = [asyncio.create_task(get_img(session, url)) for url in urls]
  
          await asyncio.wait(tasks)
  
  
  if __name__ == ‘__main__‘:
      t1 = time.time()
      asyncio.run(main())
      print(time.time()-t1)
     
  >>>>>>>>
  0.08776521682739258
  

3.异步编程

3.1 事件循环

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

# 伪代码

任务列表 = [任务1, 任务2, 任务3....]

while True:
    可执行的任务列表，已完成的任务列表 = 去任务列表中检查所有的任务，将"可执行"和"已完成"的任务返回
    for 就绪任务 in 可执行任务列表:
        执行已就绪的任务
    
    for 已完成的任务 in 已完成的任务列表:
        在任务列表中移除 已完成的任务
    
    如果 任务列表 中的任务都已完成，则终止循环

import asyncio

# 去生成或获取一个事件循环
loop = asyncio.get_event_loop()
# 将任务放到`任务列表`
loop.run_until_complete(任务)

3.2 快速上手

协程函数，定义函数时，async def 函数名。

协程对象，执行 协程函数()得到的就是协程对象。

async def func():
    pass

result = func()

注意：执行协程函数创建协程对象，函数内部代码不会执行

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

import asyncio

async def func():
    print("xxxxxx!")

result = func()

# loop = asyncio.get_event_loop()
# loop.run_until_complete(result)
asyncio.run(result)  # python3.7后

3.3 await

await + 可等待的对象（协程对象、Future、Task对象——>暂时可以理解成IO等待）

示例1：

import asyncio

async def func():
    print(‘xixixi‘)
    res = await asyncio.sleep(2)
    print("结束", res)

asyncio.run(func())

示例2：

import asyncio

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

async def func():
    print("执行协程函数内部代码")
    
    # 遇到IO操作挂起当前协程(任务)，等IO操作完成之后再继续往下执行。当前协程对象挂起时，事件循环可以去执行其他协程(任务)。
    res = await others()
    print("IO请求结束，结果为：", res)
  
asyncio.run(func())

示例3：

import asyncio

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

async def func():
    print("执行协程函数内部代码")
    
    # 遇到IO操作挂起当前协程(任务)，等IO操作完成之后再继续往下执行。当前协程对象挂起时，事件循环可以去执行其他协程(任务)。
    res1 = await others()
    print("IO请求结束，结果为：", res1)
    
    res2 = await others()
    print("IO请求结束，结果为：", res2)
  
asyncio.run(func())

await就是等待对象的值得到结果之后再继续向下走。

3.4 Task对象

在事件循环中添加多个任务的。

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

示例1：

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"


async def main():
    print("main开始")

    # 创建协程，将协程封装到一个Task对象中并立即添加到时间循环的任务列表中，等待事件循环去执行(默认是就绪状态)
    # 创建Task对象，将当前执行func函数任务添加到事件循环
    task1 = asyncio.create_task(func())
    # 创建Task对象，将当前执行func函数任务添加到事件循环
    task2 = asyncio.create_task(func())

    print("main结束")

    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务
    # 此处的await是等待相对应的协程全都执行完毕并获取结果

    res1 = await task1
    res2 = await task2
    print(res1, res2)


asyncio.run(main())
>>>>>>>>
main开始
main结束
1
1
2
2
返回值 返回值

示例2：

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"


async def main():
    print("main开始")

    task_list = [
        asyncio.create_task(func(), name=‘n1‘),
        asyncio.create_task(func(), name=‘n2‘)
    ]
    print("main结束")

    done, pending = await asyncio.wait(task_list, timeout=None)
    print(done)


asyncio.run(main())

示例3：

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"


task_list = [
    func(),
    func(), 
    ]

done,pending = asyncio.run(asyncio.wait(task_list))
print(done)

注意：asyncio.create_task是将任务添加到循环列表中，而不是创建循环列表，所有当task_list在函数外时，只能将协程对象放入task_list中

3.5.1 asyncio Future对象（理解即可）

Task继承Future,Task对象内部await结果的处理基于Future对象来的。

示例1：

import asyncio


async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()

    # 创建一个任务(Future对象)，这个任务什么都不干
    fut = loop.create_future()

    # 等待任务最终结果(Future对象)，没有结果则会一直等下去
    await fut


asyncio.run(main())

示例2：

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()

    # 创建一个任务(Task对象)，绑定了set_after函数，函数内部在2s后，会给fut赋值。
    # 即手动设置future任务的最终结果，那么fut就可以结束了。
    await loop.create_task(set_after(fut))
    
    # 等待Future对象获取 最终结果 否则一直等下去
    data = await fut
    print(data)

asyncio.run(main())

3.5.2 concurrent.futures.Future对象

使用线程池、进程池实现异步操作时用到的对象。

import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor

def func(val):
    time.sleep(1)
    print(val)

# 创建线程池
pool = ThreadPoolExecutor(max_workers=5)
# 或 pool = ProcessPoolExecutor(max_workers=5)

for i in range(10):
    fut = pool.submit(func, i)
    print(fut)

以后写代码可能会存在交叉使用。例如：crm项目80%都是基于协程异步编程 + MySQL（不支持）【线程、进程做异步编程】。

import time
import asyncio
import concurrent.futures

def func1():
    # 某个耗时操作
    time.sleep(2)
    return "SB"

async def main():
    loop = asyncio.get_running_loop()

    # 第一步:内部会先调用 ThreadPoolExecutor 的 submit 方法去线程池中申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象
    # 第二步:调用asyncio.wrap_future将concurrent.futures.Future对象包装为asyncio.Future对象。
    # 因为concurrent.futures.Future对象不支持await语法，所有需要包装为 asyncio.Future对象 才能使用
    fut = loop.run_in_executor(None, func1)
    res = await fut	
    print("default thread pool", res)

    # 2.Run in a custom thread pool:
    # with concurrent.futures,ThreadPoolExecutor() as pool:
    #     res = await loop.run_in_executor(pool, func1)
    #     print(‘custom thread pool‘,res)

    # 3.Run in a custom thread pool:
    # with concurrent.futures,ProcessPoolExecutor() as pool:
    #     res = await loop.run_in_executor(pool, func1)
    #     print(‘custom process pool‘,res)


asyncio.run(main())

案例：asyncio + 不支持异步的模块

import asyncio
import requests


async def down_img(url):
    # 发送网络请求，下载图片(遭遇网络下载图片的IO请求，自顶会切换到其他任务)
    print("开始下载：", url)
	
	# 所以遇到IO阻塞时不支持异步的模块可以开启线程来执行其他的任务，但是单个线程内仍是在阻塞中
    loop = asyncio.get_event_loop()
    # requests模块默认不支持异步操作，所有就使用线程池来配合实现了。
    future = loop.run_in_executor(None, requests.get, url)

    res = await future
    print("下载完成")
    file_name = url.split(‘,‘)[-1]
    with open(file_name, ‘wb‘) as f:
        f.write(res)


if __name__ == ‘__main__‘:
    urls = [
        "https://dss2.bdstatic.com/70cFvnSh_Q1YnxGkpoWK1HF6hhy/it/u=3228549874,2173006364&fm=26&gp=0.jpg",
        "https://dss0.bdstatic.com/70cFuHSh_Q1YnxGkpoWK1HF6hhy/it/u=1280325423,1024589167&fm=26&gp=0.jpg",
        "https://dss1.bdstatic.com/70cFuXSh_Q1YnxGkpoWK1HF6hhy/it/u=3206689113,2237998950&fm=26&gp=0.jpg"
    ]
    tasks = [down_img(url) for url in urls]

    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

3.6 异步迭代器

迭代器：在其内部实现yield方法和next方法的对象。可迭代对象：在类内部实现一个iter方法，并返回一个迭代器。

异步迭代器：实现了__aiter__()和__anext__()方法的对象，必须返回一个awaitable对象。async_for支持处理异步迭代器的

anext()方法返回的可等待对象，直到引发一个stopAsyncIteration异常，这个改动由PEP 492引入。

异步可迭代对象：可在async_for语句中被使用的对象，必须通过它的__aiter__()方法返回一个asynchronous_iterator（异步迭代器）. 这个改动由PEP 492引入。

示例：async for 不能直接写在普通方法或者暴露在外面。必须写在协程函数，任意协程函数均可。

import asyncio


class Reader:
    """自定义异步迭代器(同时也是异步可迭代对象)"""

    def __init__(self):
        self.count = 0

    async def readline(self):
        # await asyncio.sleep(1)
        self.count += 1
        if self.count == 100:
            return
        return self.count

    def __aiter__(self):
        return self

    async def __anext__(self):
        val = await self.readline()
        if val is None:
            raise StopAsyncIteration
        return val
   	
async def func():
    obj = Reader()
    # async for 必须放在协程函数里，否则报错
    async for item in obj:
        print(item)


asyncio.run(func())

3.7 异步上下文管理器

此种对象通过定义aenter()和aexit()方法来对async with语句宏的环境进程控制。

async with 会先执行类中的__aenter__方法 返回对象自己
async with 也必须放在协程函数内才能执行

import asyncio


class AsyncContextManager:
    def __init__(self, conn):
        self.conn = conn

    async def do_something(self):
        # 异步操作数据库
        return 666

    async def __aenter__(self):
        # 异步数据库连接
        self.conn = await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 异步关闭数据库链接
        await asyncio.sleep((1))



async def func():
    obj = AsyncContextManager(‘conn‘)
    async with AsyncContextManager(‘conn‘) as f:
        res = await f.do_something()
        print(res)


asyncio.run(func())

4. uvloop(不支持windows平台)

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

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolocy())

# 编写asyncio的代码，与之前写的代码一致

# 内部的事件循环自动化会变为uvloop
asyncio.run(...)

注意：一个asgi——>uvicorn内部使用的就是uvloop

5.实战案例

5.1异步redis

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

pip install aioredis

示例1：

import asyncio
import aioredis


async def execute(addr, pwd):
    print("开始执行：", addr)
    # 网络IO操作：创建redis链接
    redis = await aioredis.create_redis(address=addr, password=pwd)

    # 网络IO操作：在redis中设置哈希值car，内部在设三个键值对，即：redis={car:{k1:1,k2:2,k3:3}}
    # await redis.hmset_dict(‘car‘, field1=1, field2=2, field3=3)
    await redis.hmset_dict(‘car‘, key={‘k1‘: 1, ‘k2‘: 2, ‘k3‘: 3})

    # 网络IO操作：去redis中获取值
    res = await redis.hgetall(‘car‘, encoding=‘utf-8‘)
    print(res)

    redis.close()
    # 网络IO操作：关闭redis链接
    await redis.wait_closed()

    print("结束", addr)


asyncio.run(execute("redis://localhost:6379", "root!2345"))

示例2：

import asyncio
import aioredis


async def execute(addr, pwd):
    print("开始执行：", addr)
    # 网络IO操作：先去链接 47.93.4.197:6379,遇到IO则自动切换任务，去链接 47.93.4.198:6379
    redis = await aioredis.create_redis_pool(address=addr, password=pwd)

    # 网络IO操作：遇到IO会自动切换任务
    await redis.hmset_dict(‘car‘, key={‘k1‘: 1, ‘k2‘: 2, ‘k3‘: 3})

    # 网络IO操作：遇到IO会自动切换任务
    res = await redis.hgetall(‘car‘, encoding=‘utf-8‘)
    print(res)

    redis.close()
    # 网络IO操作：遇到IO会自动切换任务
    await redis.wait_closed()

    print("结束", addr)


task_list = [
    execute(‘redis://47.93.4.197:6379‘, ‘password‘),
    execute(‘redis://47.93.4.198:6379‘, ‘password‘),
]

asyncio.run(asyncio.wait(task_list))

5.2 异步MySQL

pip install aiomysql

示例1：

import asyncio
import aiomysql


async def execute():
    # 网络IO操作：创建MySQL
    conn = await aiomysql.connect(host=‘localhost‘, port=3306, user=‘root‘, password=‘123456‘, db=‘istudy‘)

    # 网络IO操作：创建cursor
    cursor = await conn.cursor(aiomysql.cursors.DictCursor)

    # 网络IO操作：执行sql语句
    await cursor.execute("select id, username from app01_user")

    # 网络IO操作：获取sql结果
    res = await cursor.fetchall()
    print(res)

    # 网络IO操作：关闭链接
    await cursor.close()
    conn.close()


asyncio.run(execute())

示例2：

import asyncio
import aiomysql


async def execute(host, pwd):
    print("开始", host)
    # 网络IO操作：先去链接 47.93.4.197:3306,遇到IO则自动切换任务，去链接 47.93.4.198:3306
    conn = await aiomysql.connect(host=host, port=3306, user=‘root‘, password=pwd, db=‘istudy‘)

    # 网络IO操作：遇到IO会自动切换任务
    cursor = await conn.cursor(aiomysql.cursors.DictCursor)

    # 网络IO操作：遇到IO会自动切换任务
    await cursor.execute("select id, username from app01_user")

    # 网络IO操作：遇到IO会自动切换任务
    res = await cursor.fetchall()
    print(res)

    # 网络IO操作：遇到IO会自动切换任务
    await cursor.close()
    conn.close()
    print("结束", host)


task_list = [
    execute(‘47.93.4.197‘, ‘password‘),
    execute(‘47.93.4.198‘, ‘password‘),
]

asyncio.run(asyncio.wait(task_list))

5.3 FastAPI框架

pip install fastapi
pip install uvicorn(asgi内部基于uvloop)

示例：

import asyncio
import uvicorn
import aioredis
from aioredis import Redis
from fastapi import FastAPI

app = FastAPI()

REDIS_POOL = aioredis.ConnectionsPool(‘redis://xx.xx.xxx.xx:6379‘, password=‘pwd‘, minsize=1, maxsize=10)


@app.get(‘/‘)
def index():
    """普通操作接口"""
    return {"message": "hello world"}


@app.get(‘/red‘)
async def red():
    """异步操作接口"""
    print("请求来了")

    await asyncio.sleep(3)
    # 连接池获取一个链接
    conn = await REDIS_POOL.acquire()
    redis = Redis(conn)

    # 设置值
    await redis.hmset_dict(‘car‘, key={‘k1‘: 1, ‘k2‘: 2, ‘k3‘: 3})

    # 读取值
    res = await redis.hgetall(‘car‘, encoding=‘utf-8‘)
    print(res)

    # 链接归还连接池
    REDIS_POOL.release(conn)

    return res


if __name__ == ‘__main__‘:
    uvicorn.run(‘filename:app‘, host=‘localhost‘, port=5000, log_level=‘info‘)

5.4 爬虫

import asyncio
import aiohttp

from functools import wraps
from asyncio.proactor_events import _ProactorBasePipeTransport

def silence_event_loop_closed(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            return func(self, *args, **kwargs)
        except RuntimeError as e:
            if str(e) != ‘Event loop is closed‘:
                raise
    return wrapper

_ProactorBasePipeTransport.__del__ = silence_event_loop_closed(_ProactorBasePipeTransport.__del__)


async def fetch(session, url):
    print("发送请求", url)
    async with session.get(url, verify_ssl=False) as res:
        text = await res.text()
        print("得到结果：", url, len(text))


async def main():
    async with aiohttp.ClientSession() as session:
        urls = [
            "http://python.org",
            "http://www.baidu.com",
            "http://www.pythonav.com",
        ]

        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]

        await asyncio.wait(tasks)


if __name__ == ‘__main__‘:
    asyncio.run(main())

总结

注意:在使用asyncio循环执行列表中任务时最后结束时会抛出异常

解决办法：修改asyncio.proactor_events中的_ProactorBasePipeTransport方法

from functools import wraps
from asyncio.proactor_events import _ProactorBasePipeTransport

def silence_event_loop_closed(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            return func(self, *args, **kwargs)
        except RuntimeError as e:
            if str(e) != ‘Event loop is closed‘:
                raise
    return wrapper

_ProactorBasePipeTransport.__del__ = silence_event_loop_closed(_ProactorBasePipeTransport.__del__)

最大的意义：通过一个线程利用其IO等待时间去做一些其他事情。

python 异步编程

标签：jpg   set   UNC   als   close   item   通过   状态   tap   

原文地址：https://www.cnblogs.com/HinaChan/p/14704744.html