Python协程（中）

协程嵌套

使用async可以定义协程，协程用于耗时的io操作，我们也可以封装更多的io操作过程，这样就实现了嵌套的协程，即一个协程中await了另外一个协程，如此连接起来。

import asyncio

import time

async def task(x):

    print('Waiting: ', x)

    await asyncio.sleep(x)

    return 'Done after {}s'.format(x)

async def main():

    tasks = [

        asyncio.ensure_future(task(1)),

        asyncio.ensure_future(task(2)),

        asyncio.ensure_future(task(4))

    ]

    dones, pendings = await asyncio.wait(tasks)

    for i in dones:

        print('Task ret: ', i.result())

start = time.time()

loop = asyncio.get_event_loop()

loop.run_until_complete(main())

print('Time: ', time.time() - start)

如果使用的是 asyncio.gather创建协程对象，那么await的返回值就是协程运行的结果。

results = await asyncio.gather(*tasks)

for result in results:

    print('Task ret: ', result)

不在main协程函数里处理结果，直接返回await的内容，那么最外层的run_until_complete将会返回main协程的结果。

async def task(x):

    print('Waiting: ', x)

    await asyncio.sleep(x)

    return 'Done after {}s'.format(x)

async def main():

    coroutine1 = task(1)

    coroutine2 = task(2)

    coroutine3 = task(2)

    tasks = [

        asyncio.ensure_future(coroutine1),

        asyncio.ensure_future(coroutine2),

        asyncio.ensure_future(coroutine3)

    ]

    return await asyncio.gather(*tasks)

loop = asyncio.get_event_loop()

results = loop.run_until_complete(main())

for result in results:

    print('Task ret: ', result)

或者返回使用asyncio.wait方式挂起协程。

async def task(x):

    print('Waiting: ', x)

    await asyncio.sleep(x)

    return 'Done after {}s'.format(x)

async def main():

    coroutine1 = task(1)

    coroutine2 = task(2)

    coroutine3 = task(4)

    tasks = [

        asyncio.ensure_future(coroutine1),

        asyncio.ensure_future(coroutine2),

        asyncio.ensure_future(coroutine3)

    ]

    return await asyncio.wait(tasks)

loop = asyncio.get_event_loop()

done, pending = loop.run_until_complete(main())

for task in done:

    print('Task ret: ', task.result())

也可以使用asyncio的as_completed方法

async def task(x):

    print('Waiting: ', x)

    await asyncio.sleep(x)

    return 'Done after {}s'.format(x)

async def main():

    coroutine1 = task(1)

    coroutine2 = task(2)

    coroutine3 = task(4)

    tasks = [

        asyncio.ensure_future(coroutine1),

        asyncio.ensure_future(coroutine2),

        asyncio.ensure_future(coroutine3)

    ]

    for task in asyncio.as_completed(tasks):

        result = await task

        print('Task ret: {}'.format(result))

loop = asyncio.get_event_loop()

done = loop.run_until_complete(main())

协程停止

future对象有几个状态：

Pending
Running
Done
Cancelled

创建future的时候，task为pending，事件循环调用执行的时候当然就是running，调用完毕自然就是done，如果需要停止事件循环，就需要先把task取消。可以使用asyncio.Task获取事件循环的task

import asyncio

async def task(x):

    print('Waiting: ', x)

    await asyncio.sleep(x)

    return 'Done after {}s'.format(x)

tasks = [

    asyncio.ensure_future(task(1)),

    asyncio.ensure_future(task(2)),

    asyncio.ensure_future(task(3))

]

loop = asyncio.get_event_loop()

try:

    loop.run_until_complete(asyncio.wait(tasks))

except KeyboardInterrupt as e:

    print(asyncio.Task.all_tasks())

    for task in asyncio.Task.all_tasks():

        print(task.cancel())

    loop.stop()

    loop.run_forever()

finally:

    loop.close()

启动事件循环之后，马上ctrl+c，会触发run_until_complete的执行异常 KeyBorardInterrupt。然后通过循环asyncio.Task取消future。可以看到输出如下：

Waiting:  1

Waiting:  2

Waiting:  2

{<Task pending coro=<task() running at /Users/ghost/Rsj217/python3.6/async/async-main.py:18> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x101230648>()]> cb=[_wait.<locals>._on_completion() at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/tasks.py:374]>, <Task pending coro=<do_some_work() running at /Users/ghost/Rsj217/python3.6/async/async-main.py:18> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x1032b10a8>()]> cb=[_wait.<locals>._on_completion() at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/tasks.py:374]>, <Task pending coro=<wait() running at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/tasks.py:307> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x103317d38>()]> cb=[_run_until_complete_cb() at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/base_events.py:176]>, <Task pending coro=<do_some_work() running at /Users/ghost/Rsj217/python3.6/async/async-main.py:18> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x103317be8>()]> cb=[_wait.<locals>._on_completion() at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/tasks.py:374]>}

True

True

True

True

True表示cannel成功，loop stop之后还需要再次开启事件循环，最后在close，不然还会抛出异常：



Task was destroyed but it is pending!

task: <Task pending coro=<task() done,

循环task，逐个cancel是一种方案，可是正如上面我们把task的列表封装在main函数中，main函数外进行事件循环的调用。这个时候，main相当于最外出的一个task，那么处理包装的main函数即可。

import asyncio

import time

now = lambda: time.time()

async def do_some_work(x):

    print('Waiting: ', x)

    await asyncio.sleep(x)

    return 'Done after {}s'.format(x)

async def main():

    coroutine1 = do_some_work(1)

    coroutine2 = do_some_work(2)

    coroutine3 = do_some_work(2)

    tasks = [

        asyncio.ensure_future(coroutine1),

        asyncio.ensure_future(coroutine2),

        asyncio.ensure_future(coroutine3)

    ]

    done, pending = await asyncio.wait(tasks)

    for task in done:

        print('Task ret: ', task.result())

start = now()

loop = asyncio.get_event_loop()

task = asyncio.ensure_future(main())

try:

    loop.run_until_complete(task)

except KeyboardInterrupt as e:

    print(asyncio.Task.all_tasks())

    print(asyncio.gather(*asyncio.Task.all_tasks()).cancel())

    loop.stop()

    loop.run_forever()

finally:

    loop.close()

不同线程的事件循环

很多时候，我们的事件循环用于注册协程，而有的协程需要动态的添加到事件循环中。一个简单的方式就是使用多线程。当前线程创建一个事件循环，然后在新建一个线程，在新线程中启动事件循环。当前线程不会被block。

from threading import Thread

def start_loop(loop):

    asyncio.set_event_loop(loop)

    loop.run_forever()

def more_work(x):

    print('More work {}'.format(x))

    time.sleep(x)

    print('Finished more work {}'.format(x))

start = now()

new_loop = asyncio.new_event_loop()

t = Thread(target=start_loop, args=(new_loop,))

t.start()

print('TIME: {}'.format(time.time() - start))

new_loop.call_soon_threadsafe(more_work, 6)

new_loop.call_soon_threadsafe(more_work, 3)

启动上述代码之后，当前线程不会被block，新线程中会按照顺序执行call_soon_threadsafe方法注册的more_work方法，后者因为time.sleep操作是同步阻塞的，因此运行完毕more_work需要大致6 + 3

新线程协程

def start_loop(loop):

    asyncio.set_event_loop(loop)

    loop.run_forever()

async def do_some_work(x):

    print('Waiting {}'.format(x))

    await asyncio.sleep(x)

    print('Done after {}s'.format(x))

def more_work(x):

    print('More work {}'.format(x))

    time.sleep(x)

    print('Finished more work {}'.format(x))

start = now()

new_loop = asyncio.new_event_loop()

t = Thread(target=start_loop, args=(new_loop,))

t.start()

print('TIME: {}'.format(time.time() - start))

asyncio.run_coroutine_threadsafe(do_some_work(6), new_loop)

asyncio.run_coroutine_threadsafe(do_some_work(4), new_loop)

上述的例子，主线程中创建一个new_loop，然后在另外的子线程中开启一个无限事件循环。主线程通过run_coroutine_threadsafe新注册协程对象。这样就能在子线程中进行事件循环的并发操作，同时主线程又不会被block。一共执行的时间大概在6s左右。

master-worker主从模式

对于并发任务，通常是用生成消费模型，对队列的处理可以使用类似master-worker的方式，master主要用户获取队列的msg，worker用户处理消息。

为了简单起见，并且协程更适合单线程的方式，我们的主线程用来监听队列，子线程用于处理队列。这里使用redis的队列。主线程中有一个是无限循环，用户消费队列。

while True:

        task = rcon.rpop("queue")

        if not task:

            time.sleep(1)

            continue

        asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)

给队列添加一些数据：

127.0.0.1:6379[3]> lpush queue 2

(integer) 1

127.0.0.1:6379[3]> lpush queue 5

(integer) 1

127.0.0.1:6379[3]> lpush queue 1

(integer) 1

127.0.0.1:6379[3]> lpush queue 1

可以看见输出：

Waiting  2

Done 2

Waiting  5

Waiting  1

Done 1

Waiting  1

Done 1

Done 5

我们发起了一个耗时5s的操作，然后又发起了连个1s的操作，可以看见子线程并发的执行了这几个任务，其中5s awati的时候，相继执行了1s的两个任务。

巴特西