09-5 | asyncio基本使用

第一节 关于asyncio

asyncio 在3.4 版本中加入到标准库, asyncio基于selector实现, 看似库, 其实是个框架, 包含异步IO, 事件循环, 协程, 任务等内容。

通过asyncio可以实现协程等机制。

第二节 协程

前面说到因为CPython中GIL的存在, 所有核心上, 同一时间只能运行一个线程, 遇到IO或cpu时间到时, 操作系统进行线程的切换保存, 效率低。

同时使用多进程的话, 多进程间通信代价很大。

那么有没有一种方法可以将多个任务放在一个线程中执行, 当一个任务IO阻塞了, 就去运行其他任务, 这时候不会发生线程或进程的切换保存, 同时单线程环境, 线程安全, 任务通信很方便。

这样通过单线程“程序本身”控制任务的切换, 来减少操作系统切换开销, 称之为协程。

这种方式, 并不见得性能有多高, 主要是单线程环境, 不存在线程安全问题。

而asyncio就是实现了线程的一种框架。

2.1 程序控制任务切换的简单示例

# encoding = utf-8
__author__ = "mcabana.com"

def a():
    for i in range(4):
        print('a.{}'.format(i))
        yield   # 转换成一个生成器函数

def b():
    for i in 'abcd':
        print('b.{}'.format(i))
        yield   # 转换成一个生成器函数

x = a()
y = b()
for i in range(4):      # 利用yield实现多任务交替执行
    next(x)
    next(y)

yield会将函数分割成两部分转换为一个生成器对象, 每次只执行一半, 这样就可以实现多个函数单线程交替执行, 就像是协程。

其实本质就是将多个任务放到一个大循环中去, 在这个大循环中进行任务调度, 从而实现每个任务交替执行。

这种调度不是操作系统的进程线程实现的, 而是程序本身的设计完成的。这个中调度是在用户空间调度, 不需要进入内核态, 从而大大减少操作系统开销。

这种模式, 需要两种条件：

• 需要yield函数来让出控制权, 从而支持任务切换（任务就是上面写道 a, b函数）
• 需要使用大循环, 来帮助任务切换, 从而实现交替执行

下面的asyncio就是在这个基础上不断演化完善的。

2.2 协程

• 协程不是线程也不是进程, 它是使用用户空间完成并发处理的一种调度方式
• 进程线程由操作系统完成调度, 而协程是线程内完成调度, 它不需要更多的线程, 也就没有多线程切换的开销
• 协程不是抢占式调度, 只有一个协程主动退让, 另一个协程才会调度（yield就是让）
• 协程不需要锁机制, 因为在同一个线程中执行
• 多CPU下, 可以使用多进程, 每个进程内启动给一个线程, 运行协程的方式, 既能进程并发, 又能发挥多协程在线程中的优势
• Python中的协程是基于生成器的, 拨一次执行一次, 这样就可以多个协程的切换
• asyncio.iscoroutine(obj) 判断是不是协程对象
• asyncio.iscoroutinefunction(func) 判断是不是协程函数

第三节 asyncio基本使用

3.1 事件循环

asyncio 就是基于 上面的yield交替执行实现的, 他的核心是“事件循环” 也就是上面的 for循环。可以通过asyncio.get_event_loop() 来创建一个大循环。

通过这种事件循环, 可以在一个任务IO时, 切换到另外一个任务, 这样交替执行, 来充分利用CPU时间。

另外, 可以向事件循环中添加任务, 并获取任务的Future, 也就是返回结果。

一些常用方法：

asyncio.get_event_loop()	返回一个事件循环对象, 也就是创建一个大循环, 是AbstractEventLoop
AbstractEventLoop.stop()	停下运行事件循环
AbstractEventLoop.run_forever()	事件循环一直运行, 知道stop()
AbstractEventLoop.run_until_complete(future)	传入一个future对象, 运行直至future对象运行完, 返回future的结果。参数可以是Future类或子类Task的对象。一次只能传入一个future对象, 如果要传多个, 需要使用asyncio.wait() 包装。
AbstractEventLoop.colse()	关闭事件循环
AbstractEventLoop.is_running()	判断事件循环是否还在运行
AbstractEventLoop.create_task()	使用协程对象创建任务对象
asyncio.create_task()	3.7之后, 使用asyncio直接创建任务对象

3.2 事件循环示例

通过asyncio.coroutine将一个生成器函数封装成协程函数, 协程函数一调用就是 协程对象, 然后创建一个事件循环, 将协程对象放入事件循环, 就可以运行了。

如下：

# encoding = utf-8
__author__ = "mcabana.com"

import asyncio

@asyncio.coroutine
def sleep():    # 本身是一个生成器函数, 通过@asyncio.corotine转换成一个协程函数（老语法）
    for i in range(3):
        print('~~~~~~~~~~~~')
        yield

print(asyncio.iscoroutinefunction(sleep))   # 是不是一个协程函数？
print(asyncio.iscoroutine(sleep()))  # 是不是一个协程对象？

# 创建一个事件循环
loop = asyncio.get_event_loop()

# 将协程对象"sleep()"加入到这个事件大循环中, run_until_complete() 将大循环一直跑到任务完成为止
loop.run_until_complete(sleep())

# 停止事件循环
loop.close()

通过事件循环, 就可以让迭代器自己跑起来。（不然得next拨动

3.3 获取执行结果

3.3.1 方法一：future

就是将一个协程对象包装成带未来值的future对象, 然后将这个future对象加入到大循环中, 任务执行完成后, 可以通过这个future对象获取任务的返回值。

future对象有一个子类是task, 所以封装得到的 future对象也是一个task对象。

# encoding = utf-8
__author__ = "mcabana.com"

"""
带未来值的事件循环
"""

import asyncio

@asyncio.coroutine
def sleep():    # 本身是一个生成器函数, 通过asyncio.corotine转换成一个协程函数
    for i in range(3):
        print('~~~~~~~~~~~~')
        yield
    return 'res'

# 创建事件循环
loop = asyncio.get_event_loop()     

# 将协程对象封装为future对象, 包含协程对象执行完后的返回值
# ensure_future在执行时, 会调用loop.create_task()创建一个task对象
future = asyncio.ensure_future(sleep())   

print(1, future)    # 准备状态的future
loop.run_until_complete(future)     # 将future对象放入大循环中,只能放一个future对象
# 其实在run_until+complete() 处理过程中, 会判断传入的对象是不是future对象, 如果不是会调用ensure_future()将其变为一个future对象

print(2, future)    # 结束状态的future, 包含协程对象的返回值
print(future.result())  # 获取返回值, 返回值是是个字符串类型

loop.close()

3.3.2 方法二： task

# encoding = utf-8
__author__ = "mcabana.com"

"""
使用loop.creat_task() 或 asyncio.creat_task() 替代 future
"""

import asyncio

@asyncio.coroutine
def sleep():    # 本身是一个生成器函数, 通过asyncio.corotine转换成一个协程函数
    for i in range(3):
        print('~~~~~~~~~~~~')
        yield
    return 'res'

loop = asyncio.get_event_loop()     # 创建大循环

"""
方法二
使用loop.create_task 替代ensure_future, 可以直接获取result
ensuer_futuree() 中本身会调用loop.create_task创建一个task对象, future对象也是task对象
>3.7 的版本中可以使用asyncio.create_task() 代替 loop.create_task()
"""
task = loop.create_task(sleep())    # 也是将协程对象封装为 future对象

loop.run_until_complete(task)
print('task', task.result())   # 获取执行结果

loop.close()

asyncio.create_task() 方法需要在3.7之后使用

3.4 关于回调函数

回调函数, 会在task执行完成后, 立即执行。

一个大循环中会有多个任务, 每个任务的完成时间都不同, 回调函数只是在本任务执行完, 立即执行。

如下：task执行完后, 立即获取执行结果

# encoding = utf-8
__author__ = "mcabana.com"

"""
回调
"""

import asyncio

@asyncio.coroutine
def sleep():
    for i in range(3):
        print('~~~~~~~~~~~~')
        yield
    return 'res 1000'

def bro(ts):  # 回调函数必须要传入一个参数, 就是task
    print('bro {}'.format(ts.result()))     # 获取future

loop = asyncio.get_event_loop()
task = loop.create_task(sleep())

print(1, task)

# 向task中添加回调函数, task执行完, 立即执行回调函数
task.add_done_callback(bro)

print(2, task)

loop.run_until_complete(task)
print(3, task)

loop.close()

3.5 多个task执行

使用 loop.run_until_complate() 只能传入一个future对象, 使用asyncio.wait() 将多个future对象包装成一个元组, 传入。

asyncio.wait() 其实就是迭代每个coroutine对象（a, b）, 将他们转换成future对象, 然后放到一个future集合中。

# encoding = utf-8
__author__ = "mcabana.com"

"""
多个task交替执行, 并获取每个task的执行结果
"""

import asyncio

@asyncio.coroutine
def a():
    for i in range(3):
        print('a.{}'.format(i))
        yield   # 转换成一个生成器函数
    return 'a 1000'

@asyncio.coroutine
def b():
    for i in 'abc':
        print('b.{}'.format(i))
        yield   # 转换成一个生成器函数
    return 'b 1000'

# 获取future result的回调函数
def res(ts):
    print(ts.result())

loop = asyncio.get_event_loop()

task1 = loop.create_task(a())   # task1
task2 = loop.create_task(b())   # task2

task1.add_done_callback(res)    # 添加回调函数
task2.add_done_callback(res)

# 使用asyncio.wait() 将多个task放入, wait方法会迭代所有coro对象, 并将他们封装在一个future对象中
loop.run_until_complete(asyncio.wait((task1, task2)))

loop.close()

# encoding = utf-8
__author__ = "mcabana.com"

import asyncio

@asyncio.coroutine
def a():
    for _ in 'abc':
        print('a', _)
        yield
    return 'a() return = 1000'

@asyncio.coroutine
def b():
    for i in range(7):
        print('b', i)
        yield
    return 'b() return = 2000'

def cb(future):
    print(future.result())

loop = asyncio.get_event_loop()

ts = []
for t in (a(), b()):
    t = asyncio.ensure_future(t)
    t.add_done_callback(cb)
    ts.append(t)

# ret 中保存了每个任务的状态, 及返回值
ret = loop.run_until_complete(asyncio.wait(ts))

loop.close()

3.6 async def 新语法

3.5之后的新语法, async def 直接定义协程函数, 代替@asyncio.coroutine

# encoding = utf-8
__author__ = "mcabana.com"

import asyncio

@asyncio.coroutine
def wait():
    yield

# 获取future result的回调函数
def res(ts):
    print(ts.result())

"""
3.5之后的新语法, 代替@asyncio.coroutine, 
协程函数中不能出现yield, 
可以不包含await, async
"""
async def a():
    for i in range(3):
        print('a.{}'.format(i))
        """
        await后搭配一个awaitable对象, 表示该进行IO阻塞了, 该切换协程了
        await 会在此处暂停当前协程, 去执行其他协程
        awaitable对象可以是 协程对象, 也可以是实现了 __await__()方法的对象
        """
        await wait()
    return 'a 1000'

async def b():
    for i in 'abc':
        print('b.{}'.format(i))
        await wait()    # await搭配一个awaitable对象, 表示该进行IO阻塞了, 该切换协程了
    return 'b 1000'

loop = asyncio.get_event_loop()

task1 = loop.create_task(a())
task2 = loop.create_task(b())

task1.add_done_callback(res)    # 添加回调函数
task2.add_done_callback(res)

loop.run_until_complete(asyncio.wait((task1, task2)))   # 使用asyncio.wait() 将多个task放入

loop.close()

https://www.hugbg.com/archives/2850.html