Python协程与asyncio

asyncio(解决异步io编程的一整套解决方案，它主要用于异步网络操作、并发和协程)
协程(Coroutine一种用户态的轻量级微线程,它是程序级别的，在执行过程中可以中断去执行其它的子程序，别的子程序也可以中断回来继续执行之前的子程序，无需线程上下文切换的开销)

get_event_loop：创建事件循环

run_until_complete(future)：把协程注册到事件循环上，直到它执行完

# coding:utf-8
import asyncio import time # 使用async定义一个协程
async def get_corouting(): print("start get a ...") await asyncio.sleep(2) print("end get a ...") start_time = time.time() loop = asyncio.get_event_loop() loop.run_until_complete(get_corouting()) # end_time会在run_until_complete里面的协程(get_corouting)执行完后再执行
end_time = time.time() print(end_time - start_time) 执行结果： start get a ... end get a ... 2.004028797149658

create_task(coro)：创建一个task，将协程注册到事件循环中

add_done_callback(callback)：task在返回结果前执行回调函数，它的参数是接受一个方法callback，如果这个方法需要传参数可使用partial

# coding:utf-8
import asyncio from functools import partial # 使用async定义一个协程
async def get_corouting(): print("start get a ...") await asyncio.sleep(2) return "cor result"


def callback(file, future): """ 回调执行解析一个文件 :param file: :param future: :return: """
    print("resolve a file {}".format(file)) loop = asyncio.get_event_loop() task = loop.create_task(get_corouting()) task.add_done_callback(partial(callback, "a.txt")) loop.run_until_complete(task) print(task.result()) result： start get a ... resolve a file a.txt cor result

在task返回结果前先执行一个回调函数执行解析一个文件，使用partial返回函数
asyncio.wait：与线程池的wait方法一样，可接受FIRST_COMPLETED，ALL_COMPLETED等参数，用于等待协程的完成
asyncio.gather：比wait更加高级，可以对task进行分组，并且可以批量取消task

# coding:utf-8
import time import asyncio # 使用async定义一个协程
async def get_corouting(i): print("start get a{} ...".format(i)) await asyncio.sleep(2) return "cor result" start = time.time() loop = asyncio.get_event_loop() tasks1 = [get_corouting(i) for i in range(5) if i % 2 == 0] tasks2 = [get_corouting(i) for i in range(5) if i % 2 == 1] group_tasks1 = asyncio.gather(*tasks1) group_tasks2 = asyncio.gather(*tasks2) loop.run_until_complete(asyncio.gather(group_tasks1, group_tasks2)) end = time.time() print(end - start) result： start get a4 ... start get a0 ... start get a2 ... start get a3 ... start get a1 ... 2.006136178970337

在协程里面嵌套协程：

# coding:utf-8

import asyncio async def compute(x, y): print("compute {0} + {1}".format(x, y)) await asyncio.sleep(1) return x + y async def print_sum(x, y): """ await时调用compute协程 :param x: :param y: :return: """ result = await compute(x, y) print("{0} + {1} = {2}".format(x, y, result)) # 创建task
loop = asyncio.get_event_loop() # 将协程print_sum注册到loop中
loop.run_until_complete(print_sum(1, 2)) loop.close() result: compute 1 + 2
1 + 2 = 3

在asyncio事件循环中调用非协程回调函数：

call_soon：队列中等待到下一个事件循环时会立即执行

call_later：根据延时调用的时间确定执行的顺序

call_at：在指定的时间运行回调函数 这个时间是loop里面的单调时间(loop.time())

# coding:utf-8

import asyncio


def callback(sleep_times, func_name, loop):
    print(
        "{0} time {1} loop_time {2}".format(
            func_name, sleep_times, loop.time()
        )
    )


loop = asyncio.get_event_loop()
loop.call_later(3, callback, 3, "call_later", loop)
loop.call_later(2, callback, 2, "call_later", loop)
loop.call_at(loop.time(), callback, 4, "call_at", loop)
loop.call_soon(callback, 5, "call_soon", loop)
loop.run_forever()


result:
call_soon time 5 loop_time 7580.552303919
call_at time 4 loop_time 7580.552377718
call_later time 2 loop_time 7582.554425915
call_later time 3 loop_time 7583.555097398

在这个事件循环中，call_soon最先执行，接着call_at指定的时间是loop当前时间，call_at执行，随后是call_later根据延时的时间大小执行。

使用多线程在协程中集成阻塞IO：

# coding:utf-8

import asyncio import time from concurrent.futures import ThreadPoolExecutor def get_something(i): """ 用sleep模拟阻塞 :param i: :return: """ time.sleep(i) print("get {} success".format(i)) start_time = time.time() loop = asyncio.get_event_loop() executor = ThreadPoolExecutor(10) # run_in_executor：将阻塞函数放到executor(线程池)中运行
tasks = [loop.run_in_executor(executor, get_something, i) for i in range(1, 6)] # 等待task执行完成
loop.run_until_complete(asyncio.wait(tasks)) print("run time:{}".format(time.time() - start_time)) result: get 1 success get 2 success get 3 success get 4 success get 5 success run time:5.009312391281128