asyncio_mqtt 异步持续订阅三个topic 并持续发送topic

好的，让我来给您提供一个异步持续订阅三个 topic 并持续发送 topic 的代码示例。

import asyncio
from asyncio_mqtt import Client, MqttError

async def on_message(client, topic, payload, qos, properties):
    print(f'Received message on topic "{topic}": {payload.decode()}')

async def subscribe_topics(client, topics):
    await client.subscribe([(topic, 0) for topic in topics])
    print(f'Subscribed to topics: {topics}')

async def publish_topic(client, topic, message, interval):
    while True:
        try:
            await client.publish(topic, message.encode(), qos=0)
            await asyncio.sleep(interval)
        except MqttError as e:
            print(f'MQTT message publishing error: {e}')
            await asyncio.sleep(interval)

async def main():
    # 创建 MQTT 客户端
    client = Client('localhost')
    try:
        # 连接到 MQTT 代理
        await client.connect()

        # 订阅三个 topic
        topics = ['topic1', 'topic2', 'topic3']
        await subscribe_topics(client, topics)

        # 持续发送 topic4
        topic4_message = 'Hello, world!'
        topic4_interval = 1  # 每秒发送一次
        await publish_topic(client, 'topic4', topic4_message, topic4_interval)

    except MqttError as e:
        print(f'MQTT connection error: {e}')
    finally:
        # 断开连接
        await client.disconnect()

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

在上面的代码中，我们定义了三个异步函数，`on_message` 用于处理接收到的消息，`subscribe_topics` 用于订阅三个 topic，`publish_topic` 用于持续发送 topic4。

在 `main` 函数中，我们创建了一个 MQTT 客户端，连接到 MQTT 代理，并订阅了三个 topic。然后，我们使用 `publish_topic` 函数持续发送 topic4。在异常处理中，我们捕获了 `MqttError` 异常，并进行了处理。最后，在 `main` 函数中使用 `asyncio.run` 运行异步主循环。

希望这个代码示例能够帮助您！