如何用Python实现Azure IoT Hub设备消息收发？解决单设备无法同时连接问题

我正尝试用Python编写代码向Azure IoT Hub发送温度数据，同时用另一段Python代码接收该数据，两段代码都通过命令提示符运行。但目前同一设备只能建立一个连接，无法同时实现收发，请问是否可以按下图的架构实现该功能？

发布代码 #

from azure.iot.device import IoTHubDeviceClient, Message
import random
import time

# 设备连接字符串
CONNECTION_STRING = "HostName=iothub-jein02-np-eas-ua-eztr01.azure-devices.net;DeviceId=NodeMCU;SharedAccessKey=pOkHxKaDejoJUSgUU3ZdmKqG/KV3XOcbgILSX8ZDiMk="

# 创建设备客户端实例
device_client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING)

try:
    # 连接到IoT Hub
    device_client.connect()
    time.sleep(5) 

    # 生成随机温度数据
    temperature = random.randint(0, 100)
    message_payload = { "temperature": temperature } 

    # 创建并发送消息
    message = Message(str(message_payload))
    device_client.send_message(message)
    
    print("消息已成功发送到Azure IoT Hub。")
except Exception as ex:
    print(f"发生意外错误: {ex}")
finally:
    # 断开连接
    device_client.disconnect()

订阅代码 #

from azure.iot.device import IoTHubDeviceClient, Message

# 设备连接字符串
CONNECTION_STRING = "<your-device-connection-string>"

# 消息接收回调函数
def message_received_callback(message):
    print("从Azure IoT Hub收到消息:")
    print(f"    数据: {message.data}")
    print(f"    属性: {message.custom_properties}")

# 创建设备客户端实例
device_client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING)

# 设置消息接收回调
device_client.on_message_received = message_received_callback

# 连接到IoT Hub
device_client.connect()

# 保持程序运行
input("按回车键退出...\n")

# 断开连接
device_client.disconnect()

核心限制说明 #

Azure IoT Hub对同一设备ID仅允许一个活跃连接，如果用同一设备ID启动两个独立连接，后建立的连接会强制断开前一个，因此无法通过两个独立设备客户端进程实现同时收发。

可行实现方案 #

方案1：单进程单客户端同时处理收发

将发送和接收逻辑整合到同一个Python进程中，使用同一个IoTHubDeviceClient实例，这样一个连接即可支持双向通信。示例代码如下：

from azure.iot.device import IoTHubDeviceClient, Message
import random
import time
import threading

# 设备连接字符串
CONNECTION_STRING = "HostName=iothub-jein02-np-eas-ua-eztr01.azure-devices.net;DeviceId=NodeMCU;SharedAccessKey=pOkHxKaDejoJUSgUU3ZdmKqG/KV3XOcbgILSX8ZDiMk="

# 消息接收回调
def message_received_callback(message):
    print("\n从IoT Hub收到消息:")
    print(f"    数据: {message.data}")
    print(f"    属性: {message.custom_properties}")

# 发送温度数据的线程函数
def send_temperature_data(client):
    while True:
        try:
            temperature = random.randint(0, 100)
            payload = {"temperature": temperature}
            message = Message(str(payload))
            client.send_message(message)
            print(f"\n已发送温度数据: {temperature}°C")
            time.sleep(5)  # 每5秒发送一次
        except Exception as ex:
            print(f"发送消息出错: {ex}")
            break

def main():
    # 创建设备客户端
    device_client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING)
    # 设置接收回调
    device_client.on_message_received = message_received_callback
    
    try:
        # 连接到IoT Hub
        device_client.connect()
        print("已连接到Azure IoT Hub")
        
        # 启动发送数据的线程
        send_thread = threading.Thread(target=send_temperature_data, args=(device_client,))
        send_thread.daemon = True
        send_thread.start()
        
        # 保持主线程运行，等待接收消息
        input("按回车键退出...\n")
    except Exception as ex:
        print(f"连接或运行出错: {ex}")
    finally:
        # 断开连接
        device_client.disconnect()
        print("已断开与IoT Hub的连接")

if __name__ == "__main__":
    main()

方案2：分进程，接收端使用服务端SDK

如果必须分开两个独立进程，接收端不要使用设备端SDK，改用Azure IoT Hub的服务端SDK（azure-iot-hub），通过Event Hub兼容端点接收设备上传的消息。示例接收代码（服务端）：

from azure.eventhub import EventHubConsumerClient
import json

# IoT Hub Event Hub兼容连接字符串（从IoT Hub共享访问策略中获取）
EVENT_HUB_CONNECTION_STR = "<your-event-hub-compatible-connection-string>"
EVENT_HUB_NAME = "<your-event-hub-name>"

# 消息处理回调
def on_event(partition_context, event):
    print(f"\n从分区 {partition_context.partition_id} 收到消息:")
    # 解析设备上传的消息
    message_data = json.loads(event.body_as_str())
    print(f"    温度数据: {message_data.get('temperature')}°C")
    # 更新检查点
    partition_context.update_checkpoint(event)

def main():
    client = EventHubConsumerClient.from_connection_string(
        EVENT_HUB_CONNECTION_STR,
        consumer_group="$Default",
        eventhub_name=EVENT_HUB_NAME
    )
    try:
        with client:
            client.receive(
                on_event=on_event,
                starting_position="-1"  # 从最早未读取的消息开始
            )
    except KeyboardInterrupt:
        print("接收程序已停止")

if __name__ == "__main__":
    main()

总结 #

如果要按你提供的架构图实现，推荐使用方案1，通过单客户端单进程结合多线程的方式，在一个连接内同时处理消息的发送和接收，既符合IoT Hub的连接限制，也能满足你的需求。

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