欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Python协程&asyncio&异步编程

程序员文章站 2022-07-09 19:38:43
Python协程&asyncio&异步编程 1.协程 协程是微线程,是一种用户态上下文切换技术,通过一个线程实现代码块相互切换执行 实现协程有这么几种方法: greenlet,早期的模块 yield 关键字 asyncio python3.4引入的 async、await关键字 python3.5 ......

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等待时间去做其他事情。