第三章异步编程

标签：__ 异步 await 第三章编程 print async def asyncio

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("快来搞我吧！")

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("来玩呀")
    response = await asyncio.sleep(2)
    print("结束",response)

asyncio.run( func() )

示例2：

import asyncio


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


async def func():
    print("执行协程函数内部代码")

    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）。
    response = await others()

    print("IO请求结束，结果为：", response)
    
asyncio.run( func() )

示例3：

import asyncio


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


async def func():
    print("执行协程函数内部代码")

    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）。
    response1 = await others()
    print("IO请求结束，结果为：", response1)
    
    response2 = await others()
    print("IO请求结束，结果为：", response2)
    
asyncio.run( func() )

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

3.4 Task对象

Tasks are used to schedule coroutines concurrently.

When a coroutine is wrapped into a Task with functions like asyncio.create_task() the coroutine is automatically scheduled to run soon。

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

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

注意：asyncio.create_task() 函数在 Python 3.7 中被加入。在 Python 3.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() )
	
    # 创建Task对象，将当前执行func函数任务添加到事件循环。
    task2 = asyncio.create_task( func() )

    print("main结束")

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


asyncio.run( main() )

示例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)

3.5 asyncio.Future对象

A Futureis a special low-level awaitable object that represents an eventual result of an asynchronous operation.

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

示例1：

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

# 创建线程池
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()

    # 1. Run in the default loop's executor ( 默认ThreadPoolExecutor )
    # 第一步：内部会先调用 ThreadPoolExecutor 的 submit 方法去线程池中申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象
    # 第二步：调用asyncio.wrap_future将concurrent.futures.Future对象包装为asycio.Future对象。
    # 因为concurrent.futures.Future对象不支持await语法，所以需要包装为 asycio.Future对象 才能使用。
    fut = loop.run_in_executor(None, func1)
    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_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):
    # 发送网络请求，下载图片（遇到网络下载图片的IO请求，自动化切换到其他任务）
    print("开始下载:", url)

    loop = asyncio.get_event_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)


if __name__ == '__main__':
    url_list = [
        'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
        'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
        'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
    ]

    tasks = [ download_image(url)  for url in url_list]

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

3.7 异步迭代器

什么是异步迭代器

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

什么是异步可迭代对象？

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

import asyncio

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

    def __init__(self):
        self.count = 0

    async def readline(self):
        # await asyncio.sleep(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 == None:
            raise StopAsyncIteration
        return val
    
async def func():
    obj = Reader()
    async for item in obj:
        print(item)
        
asyncio.run( func() )

3.8 异步上下文管理器

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

import asyncio


class AsyncContextManager:
	def __init__(self):
        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, tb):
        # 异步关闭数据库链接
		await asyncio.sleep(1)

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

asyncio.run( func() )

标签：__,异步,await,第三章,编程,print,async,def,asyncio
From： https://www.cnblogs.com/Tmars/p/18383547

3.1 事件循环

3.2 快速上手

3.3 await

3.4 Task对象

3.5 asyncio.Future对象

3.5 concurrent.futures.Future对象

3.7 异步迭代器

3.8 异步上下文管理器

相关文章

赞助商

阅读排行

第三章 异步编程

3.1 事件循环

3.2 快速上手

3.3 await

3.4 Task对象

3.5 asyncio.Future对象

3.5 concurrent.futures.Future对象

3.7 异步迭代器

3.8 异步上下文管理器

相关文章

赞助商

阅读排行

第三章异步编程