异步服务器监听器MQTT不起作用

Question

我正在使用 pyhton3.8 和 aiomqtt 库。下面的示例代码和错误。我不明白出了什么问题。 任务是听测试题目

import asyncio
import aiomqtt


async def listen():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        async with client.messages() as messages:
            await client.subscribe("test/#")
            async for message in messages:
                print(message.payload)


background_tasks = set()


async def main():
    loop = asyncio.get_event_loop()
    task = loop.create_task(listen())
    background_tasks.add(task)
    task.add_done_callback(background_tasks.remove)
  

asyncio.run(main())

Exception in callback AbstractEventLoop.add_reader(656, <function Cli...002409DF96EE0>)
handle: <Handle AbstractEventLoop.add_reader(656, <function Cli...002409DF96EE0>)>
Traceback (most recent call last):
  File "C:\Users\test\AppData\Local\Programs\Python\Python38\lib\asyncio\events.py", line 81, in _run
    self._context.run(self._callback, *self._args)
  File "C:\Users\test\AppData\Local\Programs\Python\Python38\lib\asyncio\events.py", line 501, in add_reader
    raise NotImplementedError
NotImplementedError
Exception in callback AbstractEventLoop.add_writer(<socket.socke', 1883)>, <function Cli...002409DFAF5E0>)
handle: <Handle AbstractEventLoop.add_writer(<socket.socke', 1883)>, <function Cli...002409DFAF5E0>)>
Traceback (most recent call last):
  File "C:\Users\test\AppData\Local\Programs\Python\Python38\lib\asyncio\events.py", line 81, in _run
    self._context.run(self._callback, *self._args)
  File "C:\Users\test\AppData\Local\Programs\Python\Python38\lib\asyncio\events.py", line 507, in add_writer
    raise NotImplementedError
NotImplementedError

请帮我解答一下

Answer 1

检查工作示例，在 Windows 上运行代码需要做什么！

import asyncio
import aiomqtt

BACKGROUND_TASKS = set()  # move it here or import it from another file


async def listen():
    # keep the code the same
    async with aiomqtt.Client("test.mosquitto.org") as client:
        async with client.messages() as messages:
            await client.subscribe("test/#")
            async for message in messages:
                print(message.payload)


if __name__ == '__main__':
    loop = asyncio.SelectorEventLoop()  # create new eventloop for Windows
    asyncio.set_event_loop(loop)  # ask asyncio to use the new loop
    task = loop.create_task(listen()) # create the listener task
    BACKGROUND_TASKS.add(task)
    task.add_done_callback(BACKGROUND_TASKS.remove)
    # you need to make loop be busy with smth or run it forever
    loop.run_forever()

输出：

b'usense' b'0.00' b'12.00' b'549' b'{"在线":0,"iccid":"898604D92623D0421798"}' b'{"在线":0,"iccid":""}' 好吧 b'0' 乙'{ "log": "GPIO 5 脉冲设置为 1 se", “插件”：1， “销”：5， “模式”：“输出”， “状态”：0 } ' b'31.4' b'0' b'0' b'26.1' b'0' b'1' b'0' b'1' b'0' b'1' b'0' b'1' b'0' b'1' b'警报-1' b'警报-2' b'警报-3' b'ALARM-4' b'警报5' 无意义 b'1' b'接力' b'0' b'0.00' b'12.00' b'19' b'SNR-ERD-4' 乙'400' b“我已断开连接” b'消息20' 你好2 b'测试' b'1' b'{"1":"Hello world","2":"欢迎来到测试连接"}' b'再次嗨' b"{'addr':'https://www.google.com','端口':1234}" b'1' b'2' 乙'{ “消息”：“你好” }' b'来自 Julia 的 Hello World，使用 Mosquitto 包装器于 2023-08-06T13:37:14.154 发送 https://github.com/denglerchr/Mosquitto.jl'