如何使用Python读取Azure IoT Hub所有IoT Edge设备的C2D消息

首先明确：云到设备（C2D）消息默认是IoT Hub定向投递给指定目标设备的，没有开箱即用的全局监听所有设备C2D消息的接口。你之前使用的azure.iot.device.aio属于设备端SDK，只能绑定单个设备身份接收对应发给它的消息，无法跨设备批量读取，要实现需求需要走「IoT Hub消息路由配置+服务端SDK消费」的方案：

步骤1：配置IoT Hub自定义消息路由 #

你需要先在Azure门户的IoT Hub控制台新增一条路由规则，把所有C2D消息同步导出到可批量消费的端点：

• 路由数据源选择Cloud to device messages
• 路由端点选择IoT Hub自带的events内置端点（本质是兼容Event Hub的消息队列，无需额外创建资源）
• 路由查询条件留空即可匹配所有C2D消息，保存后启用该路由

步骤2：Python代码消费同步的C2D消息 #

使用Azure Event Hub Python SDK消费内置端点的所有C2D消息即可，无需再使用设备端SDK。

依赖安装 #

pip install azure-eventhub azure-identity

示例代码 #

import asyncio
from azure.eventhub.aio import EventHubConsumerClient
from azure.identity.aio import DefaultAzureCredential

# 以下参数均可在Azure门户IoT Hub的「内置端点」页面直接复制获取
# 事件中心兼容端点
EVENTHUB_COMPATIBLE_ENDPOINT = "替换为你的事件中心兼容端点"
# 事件中心兼容名称
EVENTHUB_COMPATIBLE_NAME = "替换为你的事件中心兼容名称"
# 消费组，建议单独新建专用消费组，不要共用默认$Default
CONSUMER_GROUP = "$Default"

async def on_event(partition_context, event):
    # 提取C2D消息内容
    msg_content = event.body_as_str()
    # 提取消息对应的目标设备ID
    target_device_id = event.properties.get("to")
    # 提取消息自定义属性（如有）
    custom_props = event.properties
    print(f"收到发给设备 {target_device_id} 的C2D消息：{msg_content}")
    # 提交 checkpoint 避免服务重启后重复消费
    await partition_context.update_checkpoint(event)

async def main():
    credential = DefaultAzureCredential()
    client = EventHubConsumerClient(
        fully_qualified_namespace=EVENTHUB_COMPATIBLE_ENDPOINT.replace("sb://", "").rstrip("/"),
        eventhub_name=EVENTHUB_COMPATIBLE_NAME,
        consumer_group=CONSUMER_GROUP,
        credential=credential
    )

    async with client:
        # starting_position设为-1表示从最新消息开始消费，可自定义修改为历史时间点
        await client.receive(on_event=on_event, starting_position="-1")

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

注意事项 #

• 运行代码的身份需要被分配IoT Hub的IoT Hub Data Reader角色，也可以直接用IoT Hub的服务端连接字符串替换DefaultAzureCredential初始化客户端，无需配置托管身份
• 如果你需要给对应设备返回C2D消息的确认回执，需要额外调用azure-iot-hub服务端SDK的回执接口操作，无法直接在消费端完成确认

内容的提问来源于stack exchange，提问作者Atif Sayeedi