在Python中从同步回调调用异步方法

Question

下面的python脚本使用从Ruuvi标记读取温度。在同步Ruuvi回调中，我们希望调用一个异步的方法（发送消息到输出）。以下代码将在第二次调用时引发异常

RuntimeError: Event loop is closed

我怎样才能使数据多次工作

import asyncio
from azure.iot.device.aio import IoTHubModuleClient
from ruuvitag_sensor.ruuvi import RuuviTagSensor

async def main():
    device_client = IoTHubModuleClient.create_from_edge_environment()
    await device_client.connect()

    def handle_data(found_data):
        asyncio.get_event_loop().run_until_complete(device_client.send_message_to_output("some data", "ruuvi"))

    while True:
        RuuviTagSensor.get_datas(handle_data)
        time.sleep(5)

    await device_client.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

Answer 1

根据您的例外情况，由于某种原因，循环似乎是闭合的。 我认为这是由于run_until_complete函数中的handle_data导致了关闭循环的反应

因此，我建议尝试以下方法：

import asyncio
from azure.iot.device.aio import IoTHubModuleClient
from ruuvitag_sensor.ruuvi import RuuviTagSensor

async def main(main_loop):
    tasks = list()
    device_client = IoTHubModuleClient.create_from_edge_environment()
    await device_client.connect()

    def handle_data(found_data):
        nonlocal main_loop
        nonlocal tasks
        tasks.append(main_loop.create_task(device_client.send_message_to_output("some data", "ruuvi")))

    while True:
        RuuviTagSensor.get_datas(handle_data)
        # We need to wait async in order to let the tasks run
        await asyncio.sleep(5)

    # This is just an insurance that all the tasks (messages to output) completed
    await asyncio.wait(tasks, timeout=5)
    await device_client.disconnect()

if __name__ == "__main__":
    # Creating and closing the loop here
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop)
    loop.close()

---

替代（更复杂的）解决方案可以使用从队列读取并调用send_message_to_output函数的函数：

import asyncio
from azure.iot.device.aio import IoTHubModuleClient
from ruuvitag_sensor.ruuvi import RuuviTagSensor

async def main(main_loop):
    q = asyncio.Queue()
    stopping = asyncio.Event()

    device_client = IoTHubModuleClient.create_from_edge_environment()
    await device_client.connect()

    async def send_msg():
        nonlocal q
        nonlocal stopping
        nonlocal device_client
        while not stopping.is_set():
            msg, sender = await q.get()
            if msg is None and sender is None:
                break
            await device_client.send_message_to_output(msg, sender)

    def handle_data(found_data):
        nonlocal q
        nonlocal stopping
        if stopping.is_set():
            return
        q.put_nowait(("some data", "ruuvi"))

    while True:
        RuuviTagSensor.get_datas(handle_data)
        await asyncio.sleep(5)

    send_msg_task = main_loop.create_task(send_msg())

    await q.put((None, None))
    await stopping.set()
    await send_msg_task
    await device_client.disconnect()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop)
    loop.close()

这里的想法是将handle_data与send_msg分开。 通过这种方式，我设法使send_msg成为一个异步函数，现在不需要创建loop或Task