azure - 仅处理来自单个设备的 Azure IoT 中心事件

Question

我正在尝试解决部署数千个 IoT 设备、将所有事件记录到 Azure IoT 中心、然后仅能够读取由单个 deviceid 创建的事件的问题。

我一直在使用EventProcessorHost来实现类似的功能，但到目前为止我只能看到一种从all读取所有消息的方法强>设备。

读取所有消息并过滤客户端并不是一个可行的解决方案，因为可能有数百万条消息。

Answer 1

Azure IoT 中心的主要用途是将设备中的大量事件摄取到云流管道，以便进行实时分析。默认遥测路径(热方式)是通过内置事件中心，其中所有事件都临时存储在 EH 分区中。除了默认端点(事件)之外，还能够根据规则(条件)将事件消息路由到自定义端点。

请注意，自定义端点的数量限制为 10 个，规则数量限制为 100 个。如果此限制符合您的业务模型，您可以非常轻松地单独流式传输 10 个设备，如 Davis 的答案中所述.

但是，如果根据超过此限制 (10+1) 的源(设备)拆分遥测流管道，则需要使用额外的 Azure 实体(组件)。

下图展示了使用 Pub/Sub 推送模型基于设备拆分遥测流管道的解决方案。

上述解决方案基于使用自定义主题发布者将流事件转发到 Azure 事件网格。事件网格事件的事件架构是 here 。事件网格的自定义主题发布器由 Azure EventHubTrigger 函数表示，其中每个流事件都映射到事件网格事件消息，主题指示已注册的设备。

Azure 事件网格是一种 Pub/Sub 松散解耦模型，其中事件根据订阅者的订阅传递给订阅者。换句话说，如果没有匹配的传递，事件消息就会消失。

请注意，事件网格路由的能力是每个区域每秒处理 1000 万个事件。每个地区的订阅数量限制为 1000 个。

使用REST Api ，订阅可以动态创建、更新、删除等。

以下代码片段显示了将流事件映射到 EG 事件消息的 AF 实现示例。正如您所看到的，这是非常简单的实现:

运行.csx:

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


using System.Configuration;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Azure.EventGrid.Models;
using Microsoft.ServiceBus.Messaging;
using Newtonsoft.Json;

// reusable client proxy
static HttpClient client = HttpClientHelper.Client(ConfigurationManager.AppSettings["TopicEndpointEventGrid"], ConfigurationManager.AppSettings["aeg-sas-key"]);

// AF
public static async Task Run(EventData ed, TraceWriter log)
{
    log.Info($"C# Event Hub trigger function processed a message:{ed.SequenceNumber}"); 

    // fire EventGrid Custom Topic
    var egevent = new EventGridEvent()
    {
        Id = ed.SequenceNumber.ToString(),
        Subject = $"/iothub/events/{ed.SystemProperties["iothub-message-source"] ?? "?"}/{ed.SystemProperties["iothub-connection-device-id"] ?? "?"}",
        EventType = "telemetryDataInserted",
        EventTime = ed.EnqueuedTimeUtc,
        Data = new
        {
            sysproperties = ed.SystemProperties,
            properties = ed.Properties,
            body = JsonConvert.DeserializeObject(Encoding.UTF8.GetString(ed.GetBytes()))
        }
    };
    await client.PostAsJsonAsync("", new[] { egevent });  
}

// helper
class HttpClientHelper
{
    public static HttpClient Client(string address, string key)
    {      
        var client = new HttpClient() { BaseAddress = new Uri(address) };
        client.DefaultRequestHeaders.Add("aeg-sas-key", key);
        return client;      
    }
}

函数.json:

    {
      "bindings": [
        {
         "type": "eventHubTrigger",
         "name": "ed",
         "direction": "in",
         "path": "<yourEventHubName>",
         "connection": "<yourIoTHUB>",
         "consumerGroup": "<yourGroup>",
         "cardinality": "many"
        }
      ],
     "disabled": false
  }

项目.json:

{
  "frameworks": {
    "net46":{
      "dependencies": {
        "Microsoft.Azure.EventGrid": "1.1.0-preview"
      }
    }
   }
}

最后，以下屏幕片段显示了 Device1 的 AF 订阅者收到的事件网格事件消息: