本页面讨论异步 Python 的常见任务流以及使用异步 Python 时可能存在的问题。
在继续之前，请查看介绍 asyncio 的官方文档和 Fuchsia Controller 的教程。
背景
以下几项通常涉及异步代码，不一定特定于 Python：
- 异步代码不一定是多线程代码。对于与 Fuchsia 控制器相关的代码，几乎所有内容都是单线程的。
- 异步代码在循环（也称为执行器）内运行，每当出现等待时，系统就会放弃该循环。
因此，可能会导致某些问题难以调试，例如，某项后台任务打算永久运行，但又引发了异常。如果您从未等待此任务，则异常仅在日志中可见。
常见误区
未保留对任务的引用
如果您循环启动某个任务，例如：
# Don't do this!
asyncio.get_running_loop().create_task(foo_bar())
并且您未保留对返回值的引用，那么 Python 垃圾回收器可能会随时取消并删除此任务。
始终确保将您创建的任务存储在变量中，例如：
# Always store a variable!
foo_bar_task = asyncio.get_running_loop().create_task(foo_bar())
然后，确保任务的生命周期与其需求相关联。如果这是特定的测试用例，请将任务存储为测试用例类的成员，并在拆解期间取消该任务。
将阻塞代码与异步代码混合使用
在测试代码时，某些实验常常会针对某些事件抛出 sleep 语句。请勿将 asyncio.sleep 与 time.sleep 混淆。前者会返回一个协程，而后者会屏蔽整个循环。
例如：
import asyncio
import time
async def printer():
""" Prints a thing."""
print("HELLO")
async def main():
task = asyncio.get_running_loop().create_task(printer())
time.sleep(1) # THIS WILL BLOCK EXECUTION.
task.cancel()
if __name__ == "__main__":
asyncio.run(main())
以上代码不会输出任何内容。不过，如果您将 time.sleep(1) 替换为 await asyncio.sleep(1)，则会输出 HELLO（这会生成异步循环）。
FIDL 服务器错误
鉴于异步编程的性质，您的 FIDL 服务器实现可能会出现运行时错误，而您却对此毫不知情，尤其是当您的 FIDL 服务器实现连接到 Fuchsia 设备上的组件时。
遗憾的是，这种情况很难调试，而且由于 FIDL 服务器实现崩溃，因此可能只会在设备上显示为 PEER_CLOSED 错误。
一种方法（用于调试）是向任务添加回调，用于检查任务中是否发生了异常并向日志发送某些内容（甚至是对程序执行硬崩溃）。
例如：
def log_exception(task):
try:
_ = task.result()
except Exception as e:
# Can log the exception here for debugging.
task = asyncio.get_running_loop().create_task(foobar())
task.add_done_callback(log_exception)
常见任务流
同步任务
首先，建议查看 Python 文档页面，了解有关可用的同步对象的入门指南。请注意，这些对象不是线程安全的。不过，其中有些方法值得借鉴。
正在等待多个任务
您可能需要等待多项任务完成。您可以使用 asyncio.wait 完成此操作。如果您要执行类似于 select 系统调用的操作，请将 return_when 参数设置为 asyncio.FIRST_COMPLETED，例如：
done, pending = await asyncio.wait(
[task1, task2, task3],
return_when=asyncio.FIRST_COMPLETED
)
结果包含一个任务元组（在本例中为 done 和 pending）。可以像处理任何其他任务对象一样处理这些任务。因此，对于 done 对象，您可以遍历这些元素，并通过检查 result() 函数来收集结果。
在同步代码中运行异步代码
在编写代码时，大多数 Fuchsia Controller 代码都是从同步代码中调用的。为了确保任务可以在后台运行，通过 asyncio.new_event_loop() 保持循环的实例可能很有意义。
这是因为 asyncio 任务必须保持在单个循环中。asyncio.Queue 等同步对象也是如此。
如果您与之互动的代码足够简单（一次运行一个 FIDL 方法），则可以使用以下代码执行此操作：
asyncio.run_until_complete(...)
但是，如果您需要运行任务或处理同步，则需要将内容封装在包含循环的类中。只需确保它在测试或对象中，有用于关闭循环的拆解代码，例如：
class FidlHandlingClass():
def __init__(self):
self._loop = asyncio.new_event_loop()
# Can launch tasks here.
self._important_task = self._loop.create_task(foo())
def __del__(self):
self._important_task.close()
self._loop.close()
您还可以使用修饰器扩展类，以便在所有公共函数中隐式使用循环（尽管它们对调用方看起来会同步），例如：
from functools import wraps
class FidlInstance():
def __init__(self, proxy_channel):
self._loop = asyncio.new_event_loop()
self.echo_proxy = fidl.fuchsia_developer_ffx.Echo.Client(proxy_channel)
def __del__(self):
self._loop.close()
def _asyncmethod(f):
@wraps(f)
def wrapped(inst, *args, **kwargs):
coro = f(inst, *args, **kwargs)
inst._loop.run_until_complete(coro)
return wrapped
@_asyncmethod
async def echo(self, string: str):
return await self.echo_proxy.echo_string(value=string)
然后，可以使用以下代码调用上述类：
client_channel = ...
inst = FidlInstance(client_channel)
print(inst.echo("foobar"))
虽然此实例可以在非异步上下文中运行，但它仍然运行异步代码。不过，这样会使得整体的读写变得更容易。