graingert-aiorun 2018.9.1

🤔 为什么？

该 run() 函数将处理应用程序关闭序列中通常需要完成的 所有 内容。你所需要做的就是编写你的协程并运行它们。

那么 run() 函数究竟做了什么？它为asyncio应用程序执行以下标准、惯用操作

• 为给定的协程创建一个 Task（将其安排在事件循环上），

• 调用 loop.run_forever()，

• 为 SIGINT 和 SIGTERM 添加默认（和智能）信号处理器，这将停止循环；

• 并且 当 循环停止（无论是通过信号直接调用），然后它会…

• …收集所有未完成的任务，

• 使用 task.cancel() 取消它们，

• 恢复运行循环，直到所有这些任务完成，

• 等待 executor 完成关闭，

• 最后关闭循环。

所有这些内容都是模板代码，你永远不会再次编写。因此，如果你使用 aiorun，这是你需要记住的

• 从单个起始协程启动所有工作

• 当接收到关闭信号时，所有当前挂起的任务将内部抛出CancelledError异常。是否在每个协程中处理它取决于您。

• 如果您想保护协程免受取消，请参阅下方的shutdown_waits_for()。

• 尽量使执行器作业保持短暂，因为关闭过程将等待它们完成。如果您需要长时间运行的线程或进程任务，请使用专用线程/子进程，并设置daemon=True。

对于一般用途，没有太多需要了解的。aiorun有一些特殊工具，您可能在特殊情况下需要它们。这些将在下文讨论。

🖥️ 关于TCP服务器启动呢？

您会在许多在线示例中看到，对于服务器，启动发生在几个run_until_complete()阶段之后，然后才是主要的“运行”部分run_forever()。我们如何使用aiorun来处理这种情况呢？

让我们重新创建标准库文档中的echo客户端 & 服务器示例。

客户端

# echo_client.py
import asyncio
from aiorun import run

async def tcp_echo_client(message):
    # Same as original!
    reader, writer = await asyncio.open_connection('127.0.0.1', 8888)
    print('Send: %r' % message)
    writer.write(message.encode())
    data = await reader.read(100)
    print('Received: %r' % data.decode())
    print('Close the socket')
    writer.close()
    asyncio.get_event_loop().stop()  # Exit after one msg like original

message = 'Hello World!'
run(tcp_echo_client(message))

服务器

import asyncio
from aiorun import run

async def handle_echo(reader, writer):
    # Same as original!
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print("Received %r from %r" % (message, addr))
    print("Send: %r" % message)
    writer.write(data)
    await writer.drain()
    print("Close the client socket")
    writer.close()

async def main():
    server = await asyncio.start_server(handle_echo, '127.0.0.1', 8888)
    print('Serving on {}'.format(server.sockets[0].getsockname()))
    try:
        # Wait for cancellation
        while True:
            await asyncio.sleep(10)
    except asyncio.CancelledError:
        server.close()
        await server.wait_closed()

run(main())

它的工作方式与原始示例相同，只是当您在服务器实例上按下CTRL-C

$ python echo_server.py
Running forever.
Serving on ('127.0.0.1', 8888)
Received 'Hello World!' from ('127.0.0.1', 57198)
Send: 'Hello World!'
Close the client socket
^CStopping the loop
Entering shutdown phase.
Cancelling pending tasks.
Cancelling task:  <Task pending coro=[...snip...]>
Running pending tasks till complete
Waiting for executor shutdown.
Leaving. Bye!

任务收集、取消和执行器关闭都是自动发生的。

💨 您喜欢 uvloop 吗？

import asyncio, aiorun

async def main():
    <snip>

if __name__ == '__main__':
    run(main(), use_uvloop=True)

请注意，您必须自己pip install uvloop。

🛡️ 智能关闭保护

这很罕见，但有时您可能希望协程在关闭序列中不被取消中断。您会在官方文档中查找asyncio.shield()。

不幸的是，shield()在关闭场景中不起作用，因为shield()提供的保护只适用于shield()使用的特定协程被直接取消的情况。

让我解释一下：如果您执行传统的关闭序列（如aiorun内部执行），这是步骤序列

• tasks = all_tasks()，然后是

• group = gather(*tasks)，然后是

• group.cancel()

内部工作方式是shield()创建一个秘密的、内部的任务——该任务也包含在上面的all_tasks()调用中！因此，它也会像其他所有东西一样接收到取消信号。

因此，我们有一个更适合我们的shield()的替代版本：shutdown_waits_for()。如果您有一个必须在关闭序列中不能被取消的协程，只需将其包裹在shutdown_waits_for()中即可！

以下是一个示例

import asyncio
from aiorun import run, shutdown_waits_for

async def corofn():
    await asyncio.sleep(60)
    print('done!')

async def main():
    try:
        await shutdown_waits_for(corofn())
    except asyncio.CancelledError
        print('oh noes!')

run(main())

如果您在60秒内按下CTRL-C，您将立即看到打印出oh noes!，然后在60秒（从开始）后打印出done!，然后程序退出。

幕后，all_tasks()会被CTRL-C取消，除了包裹在shutdown_waits_for()调用中的那些。在这方面，它类似于asyncio.shield()，但具有特殊的适用性，适用于我们的aiorun()中的关闭场景。

请注意这一点：协程最终仍然应该在某个时刻完成。这种情况的主要用例是你不希望编写显式取消处理的短期任务。

哦，你还可以像使用 asyncio.shield() 一样使用 shutdown_waits_for()。对于这种情况，它的工作方式相同。如果你正在使用 aiorun，就没有必要使用 shield()。