本頁將討論 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 裝置上的元件時。
很遺憾，這類問題難以偵錯，而且可能只會在裝置上顯示 PEER_CLOSED 錯誤，因為您的 FIDL 伺服器實作項目會當機。
其中一種做法 (用於偵錯) 是在工作中新增回呼，檢查工作中是否發生例外狀況，並將內容傳送至記錄 (或甚至強制終止程式)。
例如：
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 控制器程式碼都是從同步程式碼呼叫。為確保工作能在背景執行，透過 fuchsia_async_extension.get_loop() 保留迴圈例項可能是不錯的做法。
這是因為 asyncio 工作必須保留在單一迴圈中。同步物件 (例如 asyncio.Queue) 也是如此。
如果互動的程式碼夠簡單 (一次執行單一 FIDL 方法)，可以使用下列項目執行這項操作：
fuchsia_async_extension.get_loop().run_until_complete(...)
但如果需要執行工作或處理同步作業，建議您將相關項目封裝在含有迴圈的類別中。請務必確認，如果這是測試或物件，有可關閉迴圈的拆除程式碼，例如：
class FidlHandlingClass():
def __init__(self):
self._loop = fuchsia_async_extension.get_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 = fuchsia_async_extension.get_loop()
self.echo_proxy = fidl_fuchsia_developer_ffx.EchoClient(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"))
雖然這個執行個體可以在非非同步環境中執行，但仍會執行非同步程式碼。不過，這可讓整體讀取和寫入作業更加輕鬆。
調整程式碼以支援非同步 Python
FIDL 是一種專為非同步作業設計的語言。不過，Mobly 等測試架構預期會使用同步 Python 函式執行測試。如果您想測試的介面主要用於同步 Python，就適用這項功能。
fuchsia-controller 的 wrappers 模組下有一些混入程式碼，可支援在同步環境中使用非同步程式碼，特別是 AsyncAdapter，可用做包裝函式或混入，以及裝飾器 asyncmethod。
搭配使用時，這會建立設定，讓特定執行個體中執行的所有程式碼，都能針對常見的 asyncio 事件迴圈執行。
在大多數情況下，您可以使用 asyncio.run()，在同步 Python 中執行非同步 Python。不過，如果您有預期會在函式呼叫中檢查狀態的佇列或非同步工作，這會變得非常困難。每次叫用 asyncio.run() 都會建立全新的事件迴圈，這可能會導致管理非同步資料結構時發生問題，因為這些結構預期一次只會在一個事件迴圈中使用。只要保留單一 asyncio 事件迴圈，即可避免發生這些例外狀況。
舉例來說，假設您要編寫 Mobly 測試案例，但您公開的類別依賴某些基礎的非同步 Python，您可以編寫如下程式碼，允許在 Mobly 中使用：
import asyncio
from mobly import asserts, base_test, test_runner
from fuchsia_controller_py.wrappers import AsyncAdapter, asyncmethod
# AsyncAdapter should be included first in the list of base classes.
class ClassYouWantToTest(AsyncAdapter, SomeBaseClass):
def __init__(self):
super().__init__()
self.async_init()
@asyncmethod
async def async_init(self):
self.queue: asyncio.Queue[int] = asyncio.Queue()
await self.queue.put(1)
@asyncmethod
async def function_we_care_about(self) -> int:
got = await self.queue.get()
self.queue.task_done()
return got
class ExampleTest(base_test.BaseTestClass):
def test_case_example(self) -> None:
"""Example... doesn't really do much useful."""
c = ClassYouWantToTest()
asserts.assert_equal(c.function_we_care_about(), 1)
if __name__ == "__main__":
test_runner.main()
方法 function_we_care_about 會經過包裝，因此執行 Mobly 測試案例時，會執行下列程式碼：
def function_we_care_about(self) -> None:
coro = self._function_we_care_about_impl()
self._mixin_asyncio_loop.run_until_complete(coro)