AsyncIO - 如何测试我的函数是否阻塞？

Question

我正在尝试测试我是否正确使用 asyncio 库（aiofiles）来查看它是否被阻塞。

我怎样才能做到这一点？我的想法是让 asyncio.sleep 无限循环检查每个调用之间的延迟，同时我尝试在两者之间插入一个阻塞调用

这是我目前的尝试，但从blocking未被调用，我假设问题是无限循环实际上也在阻塞。

from asyncio import sleep, get_event_loop
import time


async def test_blocking():
  while True:
    timea = time.time()
    await sleep(1)
    timeb = time.time() - timea
    delay = timeb - 1.00
    if delay >= 0.01:
      print(f'Function blocked by {delay}s')

async def blocking():
  print('Running blocking function')
  time.sleep(5)


loop = get_event_loop()
loop.run_until_complete(test_blocking())
loop.run_until_complete(blocking())

Answer 1

您的test_blocking协程实际上并未阻塞事件循环 - 它已正确实现。但是，run_until_complete是阻塞调用。因此，您的脚本实际上从未真正上run_until_complete(blocking())线 - 它正在等待test_blocking返回，这永远不会发生，因为它是一个无限循环。loop.create_task改为使用，您将看到您期望的行为。

此外，您可能需要考虑使用asyncio的调试模式来实现您在此处尝试执行的操作。启用后，它可以在特定回调花费的时间超过给定时间时提醒您：

记录超过 100 毫秒的回调。该 loop.slow_callback_duration属性可用于设置被视为“慢”的最短执行持续时间（以秒为单位）。

在您的测试脚本上启用时，它会打印以下内容：

Executing <Task finished name='Task-2' coro=<blocking() done, defined at a.py:13> result=None created at /home/dan/.pyenv/versions/3.8.3/lib/python3.8/asyncio/base_events.py:595> took 5.005 seconds