如何从Azure Function向Event Hub发送消息？IoT Hub触发场景

我来帮你把这个功能完整实现并理清楚细节～

你的代码框架已经搭好了，我帮你补全逻辑并说明关键要点：

完整代码示例 #

#r "Microsoft.ServiceBus"
#r "Newtonsoft.Json"

using System;
using System.Text;
using Microsoft.ServiceBus.Messaging;
using Newtonsoft.Json;

// 自定义类存放要提取的信息（根据你的实际需求调整字段）
public class ProcessedDeviceMessage
{
    public string DeviceId { get; set; }
    public DateTime EnqueuedTime { get; set; }
    public string Temperature { get; set; } // 示例：假设消息包含温度字段
}

public static void Run(EventData eventData, out string outputEventHubMessage, TraceWriter log)
{
    try
    {
        // 1. 提取IoT Hub消息的系统元数据
        var deviceId = eventData.SystemProperties["iothub-connection-device-id"].ToString();
        var enqueuedTime = eventData.SystemProperties["iothub-enqueuedtime"].ToDateTime();
        
        log.Info($"收到来自设备 {deviceId} 的消息，入队时间：{enqueuedTime}");

        // 2. 解析消息体内容（默认IoT Hub消息多为JSON格式）
        string rawMessage = Encoding.UTF8.GetString(eventData.GetBytes());
        dynamic payload = JsonConvert.DeserializeObject(rawMessage);
        
        // 提取你需要的业务字段，这里以Temperature为例，按需替换
        string temperature = payload.temperature?.ToString();

        // 3. 构造要发送到Event Hub的结构化消息
        var processedMsg = new ProcessedDeviceMessage
        {
            DeviceId = deviceId,
            EnqueuedTime = enqueuedTime,
            Temperature = temperature
        };

        // 序列化为JSON字符串作为输出
        outputEventHubMessage = JsonConvert.SerializeObject(processedMsg);
        log.Info($"已准备好要发送到Event Hub的消息：{outputEventHubMessage}");
    }
    catch (Exception ex)
    {
        log.Error($"处理消息时出错：{ex.Message}", ex);
        // 出错时返回带错误标记的消息，避免Function实例崩溃
        outputEventHubMessage = JsonConvert.SerializeObject(new 
        { 
            Error = ex.Message, 
            OriginalMessageId = eventData.MessageId 
        });
    }
}

关键配置与说明 #

• 依赖引用：添加Newtonsoft.Json是因为IoT Hub的消息通常是JSON格式，这个库是Azure Functions处理JSON的标配工具。
• 系统属性提取：IoT Hub传递的EventData里，SystemProperties字典包含了设备ID、消息入队时间这类核心元数据，直接通过键名获取即可。
• 消息体解析：先用Encoding.UTF8.GetString把二进制消息转成字符串，再反序列化为动态对象（或强类型对象），方便提取业务字段。
• 错误处理：加入try-catch块避免单个消息处理失败导致整个Function实例异常，同时记录错误日志方便排查问题。

Function配置注意事项 #

1. 输出绑定设置：在function.json里要配置Event Hub的输出绑定，示例如下：

{
  "bindings": [
    {
      "type": "eventHubTrigger",
      "name": "eventData",
      "direction": "in",
      "eventHubName": "你的IoT Hub兼容Event Hub名称",
      "connection": "IoTHubConnectionString",
      "consumerGroup": "$Default"
    },
    {
      "type": "eventHub",
      "name": "outputEventHubMessage",
      "direction": "out",
      "eventHubName": "目标Event Hub名称",
      "connection": "EventHubConnectionString"
    }
  ]
}

2. 连接字符串管理：在Azure Function的应用设置中添加IoTHubConnectionString（IoT Hub的兼容Event Hub连接字符串）和EventHubConnectionString（目标Event Hub的连接字符串），绝对不要硬编码在代码里。

自定义调整建议 #

• 如果你的消息不是JSON格式（比如纯文本/二进制），修改消息体解析的逻辑即可。
• 根据实际需要提取的字段，调整ProcessedDeviceMessage类的属性，匹配你的业务需求。
• 若需要批量处理消息，可以把输出参数改成ICollector<string>或IAsyncCollector<string>，支持一次发送多条消息。

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