如何在 Python 中异步执行后台数据库写入任务（不阻塞主线程）

本文介绍在 azure function 等有执行时间限制的环境中，如何将耗时的 cosmos db 批量写入操作移至后台线程执行，避免超时失败，并支持异常通知。核心方案是使用 `Threading.thread` 启动非阻塞任务，同时兼顾错误捕获与可观测性。

在 Python Web 或 Serverless 场景（如 Azure Functions）中，当需要向 Cosmos DB 写入大量数据时，常因单次调用耗时过长（>2 分钟）触发平台超时而失败。此时，不能简单用 asyncio 包裹同步 I/O 方法（如示例中的 storedataToDb()），因为 asyncio.run() 会阻塞等待协程完成，而该方法本身是同步阻塞的（含 time.sleep()），根本未实现真正异步——这也是原代码仍卡住的根本原因。

✅ 正确解法：使用多线程（threading.Thread）
cosmos DB 写入属于典型的 I/O 密集型操作，python 的 GIL 不会成为瓶颈，线程是轻量、安全且高效的选择：

from threading import Thread import logging  def import_and_background_task(myobj):     """启动后台线程执行 DB 写入，立即返回"""     def _safe_store():         try:             myobj.storeDataToDb()         except Exception as e:             logging.error(f"Background DB write failed: {e}")             # ✅ 此处可集成邮件发送逻辑，例如：             # send_failure_email(f"DB update failed for {myobj.id}: {str(e)}")      thread = Thread(target=_safe_store, name="CosmosDB-Writer")     thread.daemon = False  # 关键：设为非守护线程，确保其能完整执行     thread.start()     logging.info("Background DB write started. Returning response immediately.")  def executeReport():     myobj.readInputFile()     myobj.readDataFromDb()     myobj.generateReport()      # 启动后台写入 → 不等待！     import_and_background_task(myobj)      print('Response sent immediately — background task is running.')     return {"success": True}

⚠️ 注意事项与最佳实践：

• 不要用 daemon=True：守护线程会在主线程退出时被强制终止，可能导致数据写入中断或丢失。Azure Function 实例回收前主线程结束，守护线程会被杀掉。
• 错误必须显式捕获：后台线程内异常不会传播到主线程，务必在 target 函数内 try/except 并记录日志或触发告警（如发送邮件）。
• 避免共享状态竞争：确保 myobj 在线程间是线程安全的（例如不修改同一实例属性），或改用线程局部副本（copy.deepcopy(myobj)）或参数传值。
• 进阶推荐：concurrent.futures.ThreadPoolExecutor
  若需批量提交、结果回调、超时控制或任务队列，优先使用更健壮的 ThreadPoolExecutor：

from concurrent.futures import ThreadPoolExecutor import atexit  executor = ThreadPoolExecutor(max_workers=3) atexit.register(lambda: executor.shutdown(wait=False))  # 安全清理  def executeReport():     # ... 其他逻辑     future = executor.submit(myobj.storeDataToDb)  # 非阻塞提交     future.add_done_callback(lambda f: handle_db_result(f))     return {"success": True}  def handle_db_result(future):     if future.exception():         logging.error("DB write failed", exc_info=future.exception())         send_failure_email(str(future.exception()))

总结：面对 Azure Functions 的 2 分钟硬性超时，将同步 DB 写入卸载至独立线程是最直接、低侵入的解决方案。关键在于正确启线程（非 daemon）、主动捕获异常、并建立可靠的通知机制——这比强行“伪异步”更稳定、更易维护。