Title

Python协程&asyncio&异步编程

1.协程

协程是微线程，是一种用户态上下文切换技术，通过一个线程实现代码块相互切换执行
实现协程有这么几种方法：

• greenlet，早期的模块
• yield 关键字
• asyncio python3.4引入的
• async、await关键字 python3.5 主流[推荐]

1.1 greenlet实现协程

pip install greenlet

# -*- coding: utf-8 -*-
from greenlet import greenlet


def func1():
    print(1)       # 第1步：输出1
    gr2.switch()   # 第2步：跳到func2函数
    print(2)       # 第5步：输出2
    gr2.switch()   # 第6步：跳到func2函数


def func2():
    print(3)      # 第3步：输出3
    gr1.switch()  # 第4步：跳到func1函数
    print(4)      # 第7步：输出4


gr1 = greenlet(func1)
gr2 = greenlet(func2)

gr1.switch()   # 第1步：去执行func1函数

1.2 yield关键字

# -*- coding: utf-8 -*-


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

Python3.4以及之后

# -*- coding: utf-8 -*-
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及以后

# -*- coding: utf-8 -*-
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))

1.5 一般用greenlet或者async&await关键字

2.协程的意义

在一个线程中遇到IO等待时间，不让线程一直白白等待，而是让线程利用空闲时间去点其它的事

# -*- coding: utf-8 -*-
import asyncio
import aiohttp


async def fetch(session, url):
    print("发送请求")
    async with session.get(url, verify_ssl=False) as response:
        content = await response.content.read()
        file_name = url.rsplit("/")[-1]
        with open(file_name, mode="wb") as file_object:
            file_object.write(content)
        print("下载完成", url)


async def main():
    async with aiohttp.ClientSession() as session:
        url_list = [
            "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxxa59lyj30kk10p77m.jpg",
            "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxrlhlhaj30kk0dpmxj.jpg",
            "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxrlrw4tj30kk0pp78u.jpg"
        ]

        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)

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

3.异步编程

3.1 事件循环

理解成为一个死循环

"""
# 伪代码

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

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

import asyncio

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

3.2 快速上手

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

async def func():
    pass
    
result = func()

注意：运行协程函数得到协程对象，函数内部代码不会执行
要运行协程函数内部代码，必须要将协程对象交给事件循环来处理

import asyncio

async def func():
	print("Hello World")
    
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("start")
    response = await asyncio.sleep(2)
    print("end", response)
    
asyncio.run(func)

示例2:

# -*- coding: utf-8 -*-
import asyncio


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


async def func():
    print("开始执行func")
    response = await others()
    print("IO请求结束，结果为: ", response)


asyncio.run(func())

示例3:

# -*- coding: utf-8 -*-
import asyncio


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


async def func():
    print("开始执行func")
    response1 = await others()
    print("IO请求结束，结果为: ", response1)

    response2 = await others()
    print("IO请求结束，结果为: ", response2)

asyncio.run(func())

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

3.4 Task对象

在事件循环中添加多个任务
Tasks用语并发调度协程，通过asyncio.create_task(协程对象)，这样可以让协程加入事件循环中等待被调度执行。除了使用asyncio.create_task(协程对象)还可以使用较低级的loop.create_task()或ensure_future()函数。不建议手动实例化Task对象。
注意：asyncio.create_task(协程对象)是在Python3.7时加入的。在Python3.7前可以使用较低级的asyncio.ensure_future()函数。
示例1:

import asyncio


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


async def main():
    print("开始执行main")
    # 创建Task对象，将当前执行func函数任务添加到事件循环
    task1 = asyncio.create_task(func())
    task2 = asyncio.create_task(func())
    print("main结束")

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

    print(ret1, ret2)

asyncio.run(main())

示例2:

# -*- coding: utf-8 -*-
import asyncio


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


async def main():
    print("开始执行main")
    # 创建Task对象，将当前执行func函数任务添加到事件循环
    task_list = [
        asyncio.create_task(func(), name="n1"),
        asyncio.create_task(func(), name="n2")
    ]

    print("main结束")

    # 当执行协程遇到IO操作时，会自动化切换到其它任务(task2)
    # 此处的await是等待相对应的协程全部执行完毕并获取结果
    # done是集合，是上面两个任务的返回值
    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)

3.5 asyncio.Future对象

Task继承Future，Task对象内部的await结果处理基于Future对象来着的
示例1:

import asyncio


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


asyncio.run(main())

示例2:

import asyncio


async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result("success")

async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()
    # 创建一个future对象，这个对象什么都不干，没绑定任何行为，则这个任务不知道什么时候结束
    fut = asyncio.create_future()

    # 创建一个任务(Task对象)，绑定了set_after函数，函数在2s之后，会给fut赋值
    # 即手动给fut设置结果，那么fut就可以结束了
    await loop.create_task(set_after(fut))

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


asyncio.run(main())

3.6 concurrent.future.Future对象

使用线程池或进程池实现异步操作时用到的对象
交叉使用，协程异步编程 + MySQL(不支持)[线程、进程做异步编程]

import time
import asyncio
import concurrent.futures


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


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

    # 1.Run in the default loop's excutor (默认ThreadPoolExecutor)
    # 第一步：内部会先调用ThreadPoolExecutor的submit方法去线程池中申请一个线程去执行func函数，并返回一个concurrent.futures.Future对象
    # 第二步：调用asyncio.wrap_future将concurrent.futures.Future对象包装为调用asyncio.Future对象
    # 因为concurrent.futures.Future对象不支持await语法，需要包装为asyncio.Future对象才能使用
    fut = loop.run_in_excutor(None, func)
    result = await fut
    print("default thread pool", result)

    # 2.Run in a custom thread pool
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_excutor(pool, func)
        print("custom thread pool", result)

    # 3.Run in a custom process pool
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_excutor(pool, func)
        print("custom process pool", result)


asyncio.run(main())

案例：asyncio + 不支持的模块

import asyncio
import requests


async def download_image(url):
    # f发送网络请求下载图片，(遇到下载图片的网络IO请求，自动切换到其它任务)
    print("开始下载", url)
    loop = asyncio.get_running_loop()

    # requests模块不支持异步操作，所以就使用线程池来配合实现了
    future = loop.run_in_executor(None, requests.get, url)

    response = await future
    print("下载完成")

    # 图片保存到本地文件
    file_name = url.rsplit("/")[-1]
    with open(file_name, mode="wb") as file_object:
        file_object.write(response.content)


def main():

    url_list = [
        "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxxa59lyj30kk10p77m.jpg",
        "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxrlhlhaj30kk0dpmxj.jpg",
        "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxrlrw4tj30kk0pp78u.jpg"
    ]

    tasks = [download_image(url) for url in url_list]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))


if __name__ == "__main__":
    main()

3.7 异步迭代器

什么是异步迭代器
实现了__aiter__()和__anext__()方法的对象。anext__必须返回一个awaitable对象，async for 会处理异步迭代器的__anext()方法所返回的可等待对象，知道其引发一个StopAsyncIteration异常。
什么是异步可迭代对象
可在async for 中使用的对象。必须通过它的__aiter__()返回一个asynchronous iterator

import asyncio


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

    async def readline(self):
        # await asyncio.sleep(2)
        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 == None:
            raise StopAsyncIteration
        return val


async def func():
    obj = Reader()
    # async for只能写在协程函数内
    async for item in obj:
        print(item)

asyncio.run(func())

3.8 异步上下文管理器

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

import asyncio


class AsyncContexManager(object):
    def __init__(self, ):
        self.conn = conn

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

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

    async def __anext__(self):
        # 异步关闭数据库
        await asyncio.sleep(2)


async def func():
    # async with 只能在协程函数中使用
    async with AsyncContexManager() as f:
        result = await f.do_something()
        print(result)

asyncio.run(func())

4.uvloop

是asyncio的事件循环的替代方案。事件循环 > asyncio的事件循环，性能比肩go

pip install uvloop

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy)


#  编写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(address, password):
    print("开始执行", address)
    # 网络IO 创建redis连接
    redis = await aioredis.create_redis(address, password=password)
    # 网络IO 在redis中设置哈希值car，内部设置3个键值对 redis = {"car": {"key1": 1, "key2": 2, "key3": 3}}
    await redis.hmset_dict("car", key1=1, key2=2, key3=3)
    # 网络IO 去redis中获取值
    result = redis.hgetall("car", encoding="utf-8")
    print(result)

    redis.close()
    # 网络IO关闭redis连接
    await redis.wait_close()

    print("结束", address)


asyncio.run(execute("redis://127.0.0.1:6379", "123456"))

示例2:

import asyncio
import aioredis


async def execute(address, password):
    print("开始执行", address)
    # 网络IO 先去连接47.93.4.197:6379，遇到IO则自动切换任务，去连接47.93.4.198:6379
    redis = await aioredis.create_redis(address, password=password)
    # 网络IO 遇到IO会自动切换任务
    await redis.hmset_dict("car", key1=1, key2=2, key3=3)
    # 网络IO 遇到IO会自动切换任务
    result = redis.hgetall("car", encoding="utf-8")
    print(result)

    redis.close()
    # 网络IO 遇到IO会自动切换任务
    await redis.wait_close()

    print("结束", address)


task_list = [
    execute("47.93.4.197:6379", "123456"),
    execute("47.93.4.198:6379", "123456"),
]

asyncio.run(asyncio.wait(task_list))

5.1 异步操作MySQL

pip install aiomysql

示例1:

import asyncio
import aiomysql


async def execute():
    print("开始执行")
    # 网络IO 连接MySQL
    conn = await aiomysql.connect(host="127.0.0.1", port=3306, user="root", password="123456", db="mysql")

    # 网络IO 创建cursor
    cur = await conn.cursor()
    # 网络IO 执行sql
    await cur.execute("seletc name from user")
    # 网络IO 获取sql结果
    result = await cur.fetchall()
    print(result)
    # 网络IO 关闭连接
    await cur.close()
    conn.close()


asyncio.run(execute())

示例2:

import asyncio
import aiomysql


async def execute(host, password):
    print("开始执行", host)
    # 网络IO 连接MySQL先去连接47.93.41.197，遇到IO则切换去连接47.93.41.198
    conn = await aiomysql.connect(host=host, port=3306, user="root", password=password, db="mysql")

    # 网络IO 遇到IO会自动切换任务
    cur = await conn.cursor()
    # 网络IO 遇到IO会自动切换任务
    await cur.execute("seletc name from user")
    # 网络IO 遇到IO会自动切换任务
    result = await cur.fetchall()
    print(result)
    # 网络IO 遇到IO会自动切换任务
    await cur.close()
    conn.close()
    print("结束", host)


task_list = [
    execute("47.93.41.197", "123456"),
    execute("47.93.41.198", "123456")
]
asyncio.run(asyncio.wait(task_list))

5.3 FastAPI框架

安装

pip install fastapi

pip install uvicorn # (asgi 可以认为是支持异步的wsgi，内部基于uvloop)

示例：mu.py

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


app = FastAPI()
REDIS_POOL = aioredis.ConnectionsPool("redis://47.193.14.198:6379", password="123", minsize=1, maxsize=10)


@app.get("/")
def index():
    """普通操作接口"""
    return {"msg": "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", 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("mu:app", host="127.0.0.1", port=5000, log_level="info")

5.4爬虫

pip install aiohttp

import asyncio
import aiohttp


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))
        return text


async def main():
    async with aiohttp.ClientSession() as session:
        url_list = [
            "https://python.org",
            "https://www.baidu.com",
            "https://tianbaoo.github.io"
        ]

        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        done, pending = await asyncio.wait(tasks)
        print(done)

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

6.总结

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