如何在Jupyter中实现同步异步兼容的传感器读取函数？

核心思路是让同步函数自动检测运行环境：如果在异步上下文（比如Jupyter的事件循环中），则返回协程对象供await调用；如果在同步环境，则用适配后的runa()函数安全运行异步版本。

步骤1：修复runa()函数，适配已有事件循环的场景 #

修改runa()，使其先检查当前是否存在运行中的事件循环，避免asyncio.run()在已有循环时抛出RuntimeError：

import asyncio

def runa(coro):
    try:
        loop = asyncio.get_running_loop()
        if loop.is_running():
            # 已有运行中的事件循环，直接在当前循环内等待协程完成
            return loop.run_until_complete(coro)
        else:
            # 无运行中的循环，用asyncio.run创建新循环执行
            return asyncio.run(coro)
    except RuntimeError:
        # 无事件循环存在，直接用asyncio.run启动
        return asyncio.run(coro)

步骤2：改造read_sensor()为双模式函数 #

让函数自动识别调用场景，同步调用时用runa()执行异步版本，异步调用时直接返回协程：

# 原异步实现保持不变
async def read_sensor_async():
    # 这里替换为实际的异步传感器读取逻辑
    await asyncio.sleep(0.1)
    return {"temperature": 25.5}

def read_sensor():
    try:
        # 检测是否处于异步上下文（即是否被await调用）
        asyncio.get_running_loop()
        # 异步环境下，返回协程对象，让调用者用await执行
        return read_sensor_async()
    except RuntimeError:
        # 同步环境下，用适配后的runa运行异步版本
        return runa(read_sensor_async())

调用示例 #

普通Python脚本（无事件循环） #

# 同步调用
sensor_data = read_sensor()
print(sensor_data)  # 输出: {'temperature': 25.5}

# 异步调用
async def main():
    sensor_data = await read_sensor()
    print(sensor_data)

asyncio.run(main())

Jupyter Notebook（已有事件循环） #

# 同步调用（不再报错）
sensor_data = read_sensor()
print(sensor_data)

# 异步调用（直接用await）
sensor_data = await read_sensor()
print(sensor_data)

原理说明 #

• 原runa()报错是因为Jupyter启动时会创建并保持一个事件循环，而asyncio.run()会尝试创建新循环，导致冲突。修复后的runa()会复用已有循环，避免错误。
• 改造后的read_sensor()通过asyncio.get_running_loop()判断运行环境：
  • 同步环境：捕获RuntimeError（无循环时抛出），用runa()执行异步函数并返回结果。
  • 异步环境：返回协程对象，允许调用者用await执行，符合异步编程规范。

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