如何调试Python异步函数阻塞问题（Airflow Triggerer报错排查）

问题根源 #

你代码里的future.result()是同步阻塞调用——哪怕用了ThreadPoolExecutor，这个方法会直接卡住当前的async线程，直到线程中的_get_records执行完毕。Airflow的Triggerer依赖asyncio事件循环的非阻塞特性，一旦某个任务阻塞超过阈值，就会触发你看到的警告。

修复方案 #

改用asyncio.run_in_executor，这个方法专门用来把同步函数包装成可await的协程，不会阻塞事件循环。不需要手动创建ThreadPoolExecutor，asyncio会默认使用线程池（也可以自定义参数）。

修改后的代码：

async def run(self):
    loop = asyncio.get_running_loop()
    while True:
        # 用run_in_executor包装同步函数，返回可await的对象
        records = await loop.run_in_executor(None, self._get_records)
        if records:
            yield TriggerEvent(True)
        await asyncio.sleep(self.poke_interval)

def _get_records(self):
    return self.hook.get_records(self.sql)

额外调试建议 #

• 如果self.hook.get_records本身执行时间很长，优先优化SQL查询逻辑，或者增大poke_interval来减少触发频率。
• 若需要自定义线程池参数，可以提前创建ThreadPoolExecutor实例，传入run_in_executor的第一个参数：

  from concurrent.futures import ThreadPoolExecutor
  
  # 在类初始化方法中创建线程池
  def __init__(self, ...):
      self.executor = ThreadPoolExecutor(max_workers=5)
  
  async def run(self):
      loop = asyncio.get_running_loop()
      while True:
          records = await loop.run_in_executor(self.executor, self._get_records)
          # ... 后续逻辑
  

• 保持_get_records为纯同步函数，不要在里面嵌套async逻辑，避免引发额外的阻塞问题。

内容的提问来源于stack exchange，提问作者abawb