协程 asyncio 异步编程 (笔记)

1.协程

协程不是计算机提供,程序员认为创造

协程(coroutine)也是被称为微线程,是一种用户态的上下文切换技术,简而言之,其实就是通过一个线程实现代码模块相互切换执行。例如:

def func1():
    print(1)

def func2():
    print(2)

func1()
func2()
12345678
实现协程有这么几种方法
  • greenlet,早期模块
  • yield关键字
  • aysncio装饰器(py3.4)
  • async await关键字(py3.5)

1.1 grenlet实现协程

pip3 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()            # 第1步:去执行 func1 函数

1.2 yield关键字

def func1():
    yield 1             # 第1步:生成1
    yield from func2()  # 第2步:跳转到func2
    yield 4             # 第5步:生成4

def func2():
    yield 2             # 第3步:生成 2
    yield 3             # 第4步:生成 3

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

1.3 asyncio

在python3.4及之后版本
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)

tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

注意:遇到IO阻塞会自动切换

1.4 async & await 关键字

在python3.5之后出现的关键字,让代码更加简洁方便
import asyncio


async def func1():
    print(1)
    # 假设网络请求时遇到阻塞
    await asyncio.sleep(2)     # 遇到IO耗时操作,自动切换到tasks中其他任务
    print(2)

async def func2():
    print(3)
    # 假设网络请求时遇到阻塞
    await asyncio.sleep(2)      # 遇到IO耗时操作,自动切换到tasks中其他任务
    print(4)

tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

2.协程的意义

在一个线程如果遇到IO等待时间,线程不会在等待时一直不动,而是会利用空闲时间去做其他事情

案例:去下载三张网络图片(网络IO)

  • 普通方式(同步)
  • 下载时候等待下载完成后再次发送下一次下载请求
pip3 install requests
import requests


def download_image(url, name, *args, **kwargs):
    print('start downloading:', url)
    # 下载网络图片
    request = requests.get(url, *args , **kwargs)
    print('Download complete')
    # 图片保存到本地文件
    file_name = str(name)
    with open(file_name, mode='wb') as file_object:
        file_object.write(request.content)

if __name__ == '__main__':
    url_list = []                                   # 添加图片url
    for name, main_url in enumerate(url_list):
        download_image(main_url, 'image%s' % name)  #目前下载发送请求时候函数会一个一个等待
  • 携程方式(异步)
  • 下载时候发送请求,不等待下载完成就直接再一次发送请求

安装

pip3 install aiohttp

使用

import aiohttp
import asyncio


async def fetch(session, url, name):
    print('start downloading', url)
    async with session.get(url, verify_ssl=False) as response:
        content = await response.content.read()
        file_name = name
        with open(file_name, mode='wb') as file_object:
            file_object.write(content)
        print('Download complete')

async def main():
    async with aiohttp.ClientSession() as session:
        url_list = []   # 添加图片url
                        # 通过请求下载,异步全部同时下载,全部收到请求后全部安装
        task = [asyncio.create_task(fetch(session, url, name)) for name, url in enumerate(url_list)]
        await asyncio.wait(task)

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

3.异步编程

3.1事件循环

理解成为一个死循环,去监测并执行某些代码
"""
# 伪装代码

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

while True:
    可执行任务列表 = 去任务列表中检查所有任务
    已完成任务列表 = 将'可执行任务'和'已完成任务'返回

    for 就绪任务 in 可执行任务列表:
        执行以就绪任务

    for 完成任务 in 已完成任务列表:
        在任务列表中移除,完成的任务

    if 所有任务完成:
        中止循环
"""
import asyncio

"""
# 获取事件循环
loop = asyncio.get_event_loop()

# 将任务放置任务函数
loop.run_until_complete(任务)
"""

3.2快速上手

协程函数:定义函数的时候async def 函数名
协程对象:执行携程函数()得到携程对象
async def func():
    ...

# 内部代码不会执行,只是得到了一个协程对象 
result = func()
12345
注意:执行协程函数创建协程对象,函数内部代码不会执行
如果需要运行函数内部代码,必须将协程交给协程对象交给事件循环来处理
import asyncio


async def func():
    print('run')

result = func()

# 让事件循环执行协程对象
""" python 3.6 之前 
loop = asyncio.get_event_loop()
loop.run_until_complete(result)
"""

asyncio.run(result) # python 3.7 之后

3.3 await 关键字

await + 可等待的对象(协程对象, Future, Task对象 => IO等待)

示例1:

import asyncio


async def func():
    print('wait a moment')
    await asyncio.sleep(2)
    print('ok')

asyncio.run(func())

示例2:

import asyncio


async def others():
    print('wait a moment')
    await asyncio.sleep(2)
    print('end')
    return 'values'

async def func():
    # 遇到IO操作挂起当前协程(任务),等IO操作完成后字往下执行,事件函数挂起时,事件函数可以去执行其他线程
    response = await others()
    print(response)

asyncio.run(func())

示例3:

import asyncio


async def others():
    print('wait a moment')
    await asyncio.sleep(2)
    print('end')
    return 'values'

async def func():
    # 遇到IO操作挂起当前协程(任务),等IO操作完成后字往下执行,事件函数挂起时,事件函数可以去执行其他线程
    print(await others())

    print(await others())

asyncio.run(func())

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

3.4 Task对象

事件循环中添加多个任务

Task用于并发调度协程,通过asyncio.create_task(协程对象)的方式创建Task对象,这样可以加入事件

循环中等待被调度执行,除了使用asyncio.create_task()函数以外,还可以用底层

loop.create_task或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 'values'

async def main():
    print('start main')

    # 创建Task对象,当前执行func函数任务添加到事件循环
    task1 = asyncio.create_task(func())
    task2 = asyncio.create_task(func())
    print('main end')

    # 当执行协程遇到IO操作时,会自动化切换其它任务
    # 此处await的等待相对应的协程全部执行完毕并获取结果
    value1 = await task1
    value2 = await task2
    print(value1, value2)

asyncio.run(main())

示例2:

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return 'values'

async def main():
    print('start main')

    # 创建Task对象,当前执行func函数任务添加到事件循环
    tasks = [
        asyncio.create_task(func())
        asyncio.create_task(func())
    ]
    print('main end')

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

    # timeout 等待多少秒完成
    done, pending = await asyncio.wait(tasks, timeout=None)
    # 完成的任务 done
    # 未完成的任务 pending
    print(done) # 返回的是集合

asyncio.run(main())

示例:

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return 'values'


tasks = [func(), func()]
done, pending = asyncio.run(asyncio.wait(tasks))
print(done)

3.5 asyncio.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.6 concurrent.futures.Future对象

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

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


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


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

# 创建进程池
# pool = ProcessPoolExecutor(max_workers=5)


for i in range(10):
    # submit 拿出线程池一个线程执行函数
    fut = pool.submit(func, i)
    print(fut)

在以后代码后可能会存在交叉时间。例如:crm项目80%都是基于协程异步编程 + MySQL(不支持)【线程进程做异步编程】

示例:

import time
import asyncio
import concurrent.futures


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


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

    # 1. Run in the default loop's executor (默认ThreadPoolExecutor)
    # 第一步:内部会先调用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)
    result = await fut
    print('default thread loop', result)

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

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

asyncio.run(main())

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

import asyncio
import requests


async def download_image(url, name):
    # 发送网络请求,下载图片(遇到网络下载图片的IO请求,自动化切换到其他任务)
    print('开始下载', url)
    loop = asyncio.get_event_loop()
    # requests模块不支持异步操作,所以就使用线程池来配合实现
    future = loop.run_in_executor(None, requests.get, url)

    response = await future
    print('下载完成')
    # 图片保存到本地
    file_name = name
    with open(file_name, mode='wb') as file_object:
        file_object.write(response.content)


if __name__ == '__main__':
    urls = []
    tasks = [download_image(url, f'image{i}') for i, url in enumerate(urls)]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

效果于之前的asyncio和aiohttp是一样的,但所消耗的资源会更多,但这是遇到不得已的情况才使用

3.7 异步迭代器

什么是迭代器

实现了__aiter__()和__anext__()方法的对象。__anext__必须返回一个awaitable对象,async for会处理异步迭代器的__anext__()方法返回的可等待对象,直到其引发一个StopAsyncIteration异常,由PEP 492引入。

什么是异步迭代对象

在async for 语句中被使用对象。必须通过它的__aiter__()方法返回一个asyncchonous.iterator。由PEP 492引入。

import asyncio


class Reader(object):
    """ 自定义异步迭代器(同时也是异步可迭代对象) """
    def __init__(self):
        self.count = 0

    async def readline(self):
        # await asyncio.slep(1)
        self.count += 1
        if self.count == 100:
            return None
        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 main():
    obj = Reader()
    async for item in obj:
        print(item)


asyncio.run(main())

3.8 异步的上下文管理器

此中对象通过定义__aenter__()和__aexit__()方法来对async with语句中环境进行控制。
import asyncio


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

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

    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)

obj = AsyncContextManager()

async def main():
    async with AsyncContextManager() as f:
        result = await f.do_something()
        print(result)


asyncio.run(main())

4.uvloop

是asyncio的事件循环替代方案。事件循环 > 默认asyncio的事件循环。
pip3 install uvloop
import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# 编写asyncio的代码,于之前的代码一致

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

注意:一个asgi -> uvicorn 内部使用的就是nvloop

5. 实战案例

5.1 异步redis

在使用python代码操作redis时,链接/操作/断开都是网络IO
pip3 install aioredis

示例1:

import asyncio
import aioredis


async def execute(address, password):
    print('开始创建', address)
    # 网络IO操作:创建redis连接
    redis = await aioredis.create_redis(address, password)

    # 网络IO操作:在redis中设置哈希值car,内部在设置三个键值对,既:redis = {car: {key1:1, key2:2, key3:3}}
    await redis.hmset_dic('car', key1=1, key2=2, key3=3)

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

    redis.close()
    # 网络IO操作:关闭redis连接
    await redis.wait_closed()
    print('结束', address)


asyncio.run(execute('redis:[ip:host]', 'user!password'))

示例2:

import asyncio
import aioredis


async def execute(address, password):
    print('开始创建', address)
    # 网络IO操作:创建redis连接
    redis = await aioredis.create_redis(address, password)

    # 网络IO操作:在redis中设置哈希值car,内部在设置三个键值对,既:redis = {car: {key1:1, key2:2, key3:3}}
    await redis.hmset_dic('car', key1=1, key2=2, key3=3)

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

    redis.close()
    # 网络IO操作:关闭redis连接
    await redis.wait_closed()
    print('结束', address)


task_list = [
    execute('redis:[ip:host]', 'user!password'),
    execute('redis:[ip:host]', 'user!password')
]

asyncio.run(asyncio.wait(task_list))

5.1 异步MySQL

pip3 install aiomysql

示例1:

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 host, user from user')

    # 网络IO操作:获取SQL结果
    result = await cur.fetchall()

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


asyncio.run(execute())

5.3 FastAPI框架

pip3 install fastapi
pip3 install uvicorn # asgi 内部计入uvloop

示例:

普通操作IO

import uvicorn
from fastapi import FastAPI
app = FastAPI()


@app.get('/')
def index():
    """普通操作接口"""
    # 某个IO操作10s就会一直等待,接下来的请求无法进入
    return {'message': 'hello word'}


if __name__ == '__main__':
    uvicorn.run('luffy:app', host='127.0.0.1', port=5000, log_level='info')
import asyncio
import uvicorn
import aioredis
from aioredis import Redis
from fastapi import FastAPI
app = FastAPI()
# 创建 redis 的连接池 
REDIS_POOL = aioredis.ConnectionsPool('redis:[ip:port]', password='root123', minsize=1, maxsize=10)

@app.get('/')
def index():
    """普通操作接口"""
    # 某个IO操作10s就会一直等待,接下来的请求无法进入
    return {'message': 'hello word'}


@app.get('/red'):
async def red():
    """异步接口"""
    print('接收到请求')

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

    # 设置值
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)

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

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

if __name__ == '__main__':
    uvicorn.run(' python_file_name:app', host='127.0.0.1', port=5000, log_level='info')

5.4 异步爬虫

pip3 install aiohttp
import aiohttp
import asyncio


async def fetch(session, url):
    print('发送请求', url)
    async with session.get(url, verify_ssl=False) as response:
        text = await response.text()
        print('已经获取:', url, len(text))


async def main():
    async with aiohttp.ClientSession() as session:
        url_list = []
        tasks = [asyncio.create_task(fetch(session, url) for url in url_list)]
        done, pending = await asyncio.wait(tasks)

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

总结

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

版权声明:除特别注明外,本站所有文章均为王晨曦个人站点原创

转载请注明:出处来自王晨曦个人站点 » 协程 asyncio 异步编程 (笔记)

点赞

发表评论

电子邮件地址不会被公开。 必填项已用*标注