Azure Functions 的 Azure IoT 中心触发器

项目
06/16/2023

本文介绍如何使用 IoT 中心的 Azure Functions 绑定。 IoT 中心支持基于 Azure 事件中心绑定。

有关设置和配置详细信息，请参阅概述。

重要

虽然下述代码示例使用事件中心 API，但给定的语法适用于 IoT 中心函数。

使用函数触发器来响应发送到事件中心事件流的事件。若要设置触发器，必须具有基础事件中心的读取访问权限。触发函数时，传递给函数的消息充当字符串类型。

面向消耗计划和高级计划的事件中心缩放决策是通过基于目标的缩放完成的。有关详细信息，请参阅基于目标的缩放。

有关 Azure Functions 如何使用触发器响应发送到事件中心事件流的事件的信息，请参阅将事件中心与 Azure 上的无服务器函数集成。

Azure Functions 支持两种 Python 编程模型。定义绑定的方式取决于选择的编程模型。

使用 Python v2 编程模型，可以直接在 Python 函数代码中使用修饰器定义绑定。有关详细信息，请参阅 Python 开发人员指南。

本文同时支持两个编程模型。

重要

Python v2 编程模型目前以预览版提供。

示例

以下示例显示了用于记录事件中心触发器消息正文的 C# 函数。

[FunctionName("EventHubTriggerCSharp")]
public void Run([EventHubTrigger("samples-workitems", Connection = "EventHubConnectionAppSetting")] string myEventHubMessage, ILogger log)
{
    log.LogInformation($"C# function triggered to process a message: {myEventHubMessage}");
}

若要在函数代码中访问事件元数据，请绑定到 EventData 对象。此外，还可以通过在方法签名中使用绑定表达式来访问相同的属性。以下示例演示了获取相同数据的两种方法：

[FunctionName("EventHubTriggerCSharp")]
public void Run(
    [EventHubTrigger("samples-workitems", Connection = "EventHubConnectionAppSetting")] EventData myEventHubMessage,
    DateTime enqueuedTimeUtc,
    Int64 sequenceNumber,
    string offset,
    ILogger log)
{
    log.LogInformation($"Event: {Encoding.UTF8.GetString(myEventHubMessage.Body)}");
    // Metadata accessed by binding to EventData
    log.LogInformation($"EnqueuedTimeUtc={myEventHubMessage.SystemProperties.EnqueuedTimeUtc}");
    log.LogInformation($"SequenceNumber={myEventHubMessage.SystemProperties.SequenceNumber}");
    log.LogInformation($"Offset={myEventHubMessage.SystemProperties.Offset}");
    // Metadata accessed by using binding expressions in method parameters
    log.LogInformation($"EnqueuedTimeUtc={enqueuedTimeUtc}");
    log.LogInformation($"SequenceNumber={sequenceNumber}");
    log.LogInformation($"Offset={offset}");
}

若要成批接收事件，请将 string 或 EventData 设为数组。

注意

成批接收事件时，不能像上述示例那样使用 DateTime enqueuedTimeUtc 绑定到方法参数，且必须从每个 EventData 对象中接收这些事件

[FunctionName("EventHubTriggerCSharp")]
public void Run([EventHubTrigger("samples-workitems", Connection = "EventHubConnectionAppSetting")] EventData[] eventHubMessages, ILogger log)
{
    foreach (var message in eventHubMessages)
    {
        log.LogInformation($"C# function triggered to process a message: {Encoding.UTF8.GetString(message.Body)}");
        log.LogInformation($"EnqueuedTimeUtc={message.SystemProperties.EnqueuedTimeUtc}");
    }
}

以下示例演示了一个 C# 函数，该函数基于事件中心而触发，其中输入消息字符串会写入到日志中：

        [Function("EventHubsFunction")]
        [EventHubOutput("dest", Connection = "EventHubConnectionAppSetting")]
        public static string Run([EventHubTrigger("src", Connection = "EventHubConnectionAppSetting")] string[] input,
            FunctionContext context)
        {
            var logger = context.GetLogger("EventHubsFunction");

            logger.LogInformation($"First Event Hubs triggered message: {input[0]}");

            var message = $"Output message created at {DateTime.Now}";
            return message;
        }

以下示例演示了 function.json 文件中的一个事件中心触发器绑定以及使用该绑定的一个 C# 脚本函数。该函数记录事件中心触发器的消息正文。

以下示例显示了 Functions 运行时版本 2.x 及更高版本的 function.json 文件中的事件中心绑定数据。

{
  "type": "eventHubTrigger",
  "name": "myEventHubMessage",
  "direction": "in",
  "eventHubName": "MyEventHub",
  "connection": "myEventHubReadConnectionAppSetting"
}

C# 脚本代码如下所示：

using System;

public static void Run(string myEventHubMessage, TraceWriter log)
{
    log.Info($"C# function triggered to process a message: {myEventHubMessage}");
}

#r "Microsoft.Azure.EventHubs"

using System.Text;
using System;
using Microsoft.ServiceBus.Messaging;
using Microsoft.Azure.EventHubs;

public void Run(EventData myEventHubMessage,
    DateTime enqueuedTimeUtc,
    Int64 sequenceNumber,
    string offset,
    TraceWriter log)
{
    log.Info($"Event: {Encoding.UTF8.GetString(myEventHubMessage.Body)}");
    log.Info($"EnqueuedTimeUtc={myEventHubMessage.SystemProperties.EnqueuedTimeUtc}");
    log.Info($"SequenceNumber={myEventHubMessage.SystemProperties.SequenceNumber}");
    log.Info($"Offset={myEventHubMessage.SystemProperties.Offset}");

    // Metadata accessed by using binding expressions
    log.Info($"EnqueuedTimeUtc={enqueuedTimeUtc}");
    log.Info($"SequenceNumber={sequenceNumber}");
    log.Info($"Offset={offset}");
}

若要成批接收事件，请将 string 或 EventData 设为数组：

public static void Run(string[] eventHubMessages, TraceWriter log)
{
    foreach (var message in eventHubMessages)
    {
        log.Info($"C# function triggered to process a message: {message}");
    }
}

以下示例演示了 function.json 文件中的一个事件中心触发器绑定以及使用该绑定的 JavaScript 函数。此函数将读取事件元数据并记录消息。

以下示例显示了 function.json 文件中的事件中心绑定数据，与更高版本相比，这对于 Functions 运行时版本 1.x 而言是不同的。

Functions 2.x+
Functions 1.x

{
  "type": "eventHubTrigger",
  "name": "myEventHubMessage",
  "direction": "in",
  "eventHubName": "MyEventHub",
  "connection": "myEventHubReadConnectionAppSetting"
}

{
  "type": "eventHubTrigger",
  "name": "myEventHubMessage",
  "direction": "in",
  "path": "MyEventHub",
  "connection": "myEventHubReadConnectionAppSetting"
}

JavaScript 代码如下所示：

module.exports = function (context, myEventHubMessage) {
    context.log('Function triggered to process a message: ', myEventHubMessage);
    context.log('EnqueuedTimeUtc =', context.bindingData.enqueuedTimeUtc);
    context.log('SequenceNumber =', context.bindingData.sequenceNumber);
    context.log('Offset =', context.bindingData.offset);

    context.done();
};

若要批量接收事件，请将 function.json 文件中的 cardinality 设为 many，如以下示例所示。

Functions 2.x+
Functions 1.x

{
  "type": "eventHubTrigger",
  "name": "eventHubMessages",
  "direction": "in",
  "eventHubName": "MyEventHub",
  "cardinality": "many",
  "connection": "myEventHubReadConnectionAppSetting"
}

{
  "type": "eventHubTrigger",
  "name": "eventHubMessages",
  "direction": "in",
  "path": "MyEventHub",
  "cardinality": "many",
  "connection": "myEventHubReadConnectionAppSetting"
}

JavaScript 代码如下所示：

module.exports = function (context, eventHubMessages) {
    context.log(`JavaScript eventhub trigger function called for message array ${eventHubMessages}`);

    eventHubMessages.forEach((message, index) => {
        context.log(`Processed message ${message}`);
        context.log(`EnqueuedTimeUtc = ${context.bindingData.enqueuedTimeUtcArray[index]}`);
        context.log(`SequenceNumber = ${context.bindingData.sequenceNumberArray[index]}`);
        context.log(`Offset = ${context.bindingData.offsetArray[index]}`);
    });

    context.done();
};

下面是 PowerShell 代码：

param($eventHubMessages, $TriggerMetadata)

Write-Host "PowerShell eventhub trigger function called for message array: $eventHubMessages"

$eventHubMessages | ForEach-Object { Write-Host "Processed message: $_" }

以下示例演示了一个事件中心触发器绑定以及使用该绑定的 Python 函数。此函数将读取事件元数据并记录消息。该示例取决于是使用 v1 还是 v2 Python 编程模型。

import logging
import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="EventHubTrigger1")
@app.event_hub_message_trigger(arg_name="myhub", 
                               event_hub_name="<EVENT_HUB_NAME>",
                               connection="<CONNECTION_SETTING>") 
def test_function(myhub: func.EventHubEvent):
    logging.info('Python EventHub trigger processed an event: %s',
                myhub.get_body().decode('utf-8'))

以下示例显示了 function.json 文件中的事件中心绑定数据。

{
  "type": "eventHubTrigger",
  "name": "event",
  "direction": "in",
  "eventHubName": "MyEventHub",
  "connection": "myEventHubReadConnectionAppSetting"
}

下面是 Python 代码：

import logging
import azure.functions as func


def main(event: func.EventHubEvent):
    logging.info(f'Function triggered to process a message: {event.get_body().decode()}')
    logging.info(f'  EnqueuedTimeUtc = {event.enqueued_time}')
    logging.info(f'  SequenceNumber = {event.sequence_number}')
    logging.info(f'  Offset = {event.offset}')

    # Metadata
    for key in event.metadata:
        logging.info(f'Metadata: {key} = {event.metadata[key]}')

以下示例演示了一个记录事件中心触发器消息正文的事件中心触发器绑定。

@FunctionName("ehprocessor")
public void eventHubProcessor(
  @EventHubTrigger(name = "msg",
                  eventHubName = "myeventhubname",
                  connection = "myconnvarname") String message,
       final ExecutionContext context )
       {
          context.getLogger().info(message);
 }

在 Java 函数运行时库中，对其值来自事件中心的参数使用 EventHubTrigger 注释。带有这些注释的参数会导致函数在事件到达时运行。可以将此注释与本机 Java 类型、POJO 或使用了 Optional<T> 的可为 null 的值一起使用。

以下示例演示了广泛使用 SystemProperties 和其他绑定选项以进一步自检事件，并提供格式标准的 BlobOutput 路径（日期分层）。

package com.example;
import java.util.Map;
import java.time.ZonedDateTime;

import com.microsoft.azure.functions.annotation.*;
import com.microsoft.azure.functions.*;

/**
 * Azure Functions with Event Hub trigger.
 * and Blob Output using date in path along with message partition ID
 * and message sequence number from EventHub Trigger Properties
 */
public class EventHubReceiver {

    @FunctionName("EventHubReceiver")
    @StorageAccount("bloboutput")

    public void run(
            @EventHubTrigger(name = "message",
                eventHubName = "%eventhub%",
                consumerGroup = "%consumergroup%",
                connection = "eventhubconnection",
                cardinality = Cardinality.ONE)
            String message,

            final ExecutionContext context,

            @BindingName("Properties") Map<String, Object> properties,
            @BindingName("SystemProperties") Map<String, Object> systemProperties,
            @BindingName("PartitionContext") Map<String, Object> partitionContext,
            @BindingName("EnqueuedTimeUtc") Object enqueuedTimeUtc,

            @BlobOutput(
                name = "outputItem",
                path = "iotevents/{datetime:yy}/{datetime:MM}/{datetime:dd}/{datetime:HH}/" +
                       "{datetime:mm}/{PartitionContext.PartitionId}/{SystemProperties.SequenceNumber}.json")
            OutputBinding<String> outputItem) {

        var et = ZonedDateTime.parse(enqueuedTimeUtc + "Z"); // needed as the UTC time presented does not have a TZ
                                                             // indicator
        context.getLogger().info("Event hub message received: " + message + ", properties: " + properties);
        context.getLogger().info("Properties: " + properties);
        context.getLogger().info("System Properties: " + systemProperties);
        context.getLogger().info("partitionContext: " + partitionContext);
        context.getLogger().info("EnqueuedTimeUtc: " + et);

        outputItem.setValue(message);
    }
}

特性

进程内和独立工作进程 C# 库都使用特性来配置触发器。 C# 脚本改为使用 function.json 配置文件。

在 C# 类库中，使用 EventHubTriggerAttribute，它支持以下属性。

参数	说明
EventHubName	事件中心的名称。当事件中心名称也出现在连接字符串中时，该值会在运行时覆盖此属性。可以在应用设置中引用，例如 `%eventHubName%`
ConsumerGroup	一个可选属性，用于设置使用者组，该组用于订阅事件中心中的事件。省略时，将使用 `$Default` 使用者组。
Connection	指定如何连接到事件中心的应用设置或设置集合的名称。若要了解详细信息，请参阅连接。

使用 EventHubTriggerAttribute 在事件中心上定义触发器，EventHubTriggerAttribute 支持以下属性。

参数	说明
EventHubName	事件中心的名称。当事件中心名称也出现在连接字符串中时，该值会在运行时覆盖此属性。可以在应用设置中引用，例如 `%eventHubName%`
ConsumerGroup	一个可选属性，用于设置使用者组，该组用于订阅事件中心中的事件。省略时，将使用 `$Default` 使用者组。
Connection	指定如何连接到事件中心的应用设置或设置集合的名称。若要了解详细信息，请参阅连接。

function.json 属性	说明
type	必须设置为 `eventHubTrigger`。在 Azure 门户中创建触发器时，会自动设置此属性。
direction	必须设置为 `in`。在 Azure 门户中创建触发器时，会自动设置此属性。
name	在函数代码中表示事件项的变量的名称。
eventHubName	Functions 2.x 及更高版本。事件中心的名称。当事件中心名称也出现在连接字符串中时，该值会在运行时覆盖此属性。可以通过应用设置 `%eventHubName%` 进行引用。在版本 1.x 中，此属性名为 `path`。
consumerGroup	一个可选属性，用于设置使用者组，该组用于订阅事件中心中的事件。如果将其省略，则会使用 `$Default` 使用者组。
连接	指定如何连接到事件中心的应用设置或设置集合的名称。请参阅连接。

修饰符

仅适用于 Python v2 编程模型。

对于使用修饰器定义的 Python v2 功能，支持 cosmos_db_trigger 上的以下属性：

properties	说明
`arg_name`	在函数代码中表示事件项的变量的名称。
`event_hub_name`	事件中心的名称。当事件中心名称也出现在连接字符串中时，该值会在运行时覆盖此属性。
`connection`	指定如何连接到事件中心的应用设置或设置集合的名称。请参阅连接。

对于使用 function.json 定义的 Python 函数，请参阅“配置”部分。

批注

在 Java 函数运行时库中，使用支持以下设置的 EventHubTrigger 注释：

配置

仅适用于 Python v1 编程模型。

下表说明了在 function.json 文件中设置的触发器配置属性，这些属性因运行时版本而异。

Functions 2.x+
Functions 1.x

function.json 属性	说明
type	必须设置为 `eventHubTrigger`。在 Azure 门户中创建触发器时，会自动设置此属性。
direction	必须设置为 `in`。在 Azure 门户中创建触发器时，会自动设置此属性。
name	在函数代码中表示事件项的变量的名称。
eventHubName	事件中心的名称。当事件中心名称也出现在连接字符串中时，该值会在运行时覆盖此属性。可以通过应用设置 `%eventHubName%` 进行引用
consumerGroup	一个可选属性，用于设置使用者组，该组用于订阅事件中心中的事件。如果将其省略，则会使用 `$Default` 使用者组。
基数	设为 `many` 以启用批处理。如果省略或设为 `one`，将向函数传递一条消息。
连接	指定如何连接到事件中心的应用设置或设置集合的名称。请参阅连接。

function.json 属性	说明
type	必须设置为 `eventHubTrigger`。在 Azure 门户中创建触发器时，会自动设置此属性。
direction	必须设置为 `in`。在 Azure 门户中创建触发器时，会自动设置此属性。
name	在函数代码中表示事件项的变量的名称。
路径	事件中心的名称。当事件中心名称也出现在连接字符串中时，该值会在运行时覆盖此属性。
consumerGroup	一个可选属性，用于设置使用者组，该组用于订阅事件中心中的事件。如果将其省略，则会使用 `$Default` 使用者组。
基数	设为 `many` 以启用批处理。如果省略或设为 `one`，将向函数传递一条消息。
连接	指定如何连接到事件中心的应用设置或设置集合的名称。请参阅连接。

在本地开发时，需要将应用程序设置添加到 Values 集合中的 local.settings.json 文件中。

使用情况

要详细了解事件中心触发器和 IoT 中心触发器的缩放方式，请参阅通过 Azure Functions 使用事件。

事件中心输出绑定支持的参数类型取决于所用的 Functions 运行时版本、扩展包版本以及 C# 模态。

进程内 C# 类库函数支持以下类型：

Azure.Messaging.EventHubs.EventData
字符串
Byte Array
普通旧 CLR 对象 (POCO)

此版本的 EventData 为了支持 EventBody 舍弃了对旧式 Body 类型的支持。

参数类型可以为以下类型之一：

任意的本机 Java 类型，例如 int、String、byte[]。
使用可选的可为 null 的值。
任意 POJO 类型。

若要了解详细信息，请参阅 EventHubTrigger 参考。

事件元数据

事件中心触发器提供了几个元数据属性。元数据属性可在其他绑定中用作绑定表达式的一部分，或者用作代码中的参数。属性来自 EventData 类。

属性	类型	说明
`PartitionContext`	PartitionContext	`PartitionContext` 实例。
`EnqueuedTimeUtc`	`DateTime`	排队时间 (UTC)。
`Offset`	`string`	数据相对于事件中心分区流的偏移量。偏移量是事件中心流中的事件的标记或标识符。该标识符在事件中心流的分区中是惟一的。
`PartitionKey`	`string`	事件数据应该发送到的分区。
`Properties`	`IDictionary<String,Object>`	事件数据的用户属性。
`SequenceNumber`	`Int64`	事件的逻辑序列号。
`SystemProperties`	`IDictionary<String,Object>`	系统属性，包括事件数据。

请参阅在本文的前面部分使用这些属性的代码示例。

连接

connection 属性是对包含应用程序设置（其中包含连接字符串）名称的环境配置的引用。可以通过选择命名空间对应的“连接信息”按钮来获取此连接字符串。连接字符串必须用于事件中心命名空间，而不是事件中心本身。

该连接字符串必须至少拥有“读取”权限才可激活函数。

此连接字符串应存储在应用程序设置中，其名称与绑定配置的 connection 属性指定的值匹配。

注意

IoT 中心触发器不支持基于标识的连接。如果需要端到端使用托管标识，可以改用 IoT 中心路由将数据发送到你控制的事件中心。这样，便可以使用托管标识对出站路由进行身份验证，并可以使用托管标识从该事件中心读取事件。

host.json 属性

host.json 文件包含控制事件中心触发器行为的设置。有关可用设置的详细信息，请参阅 host.json 设置部分。

后续步骤

将事件写入事件流（输出绑定）

通过

Azure Functions 的 Azure IoT 中心触发器

示例

特性

修饰符

批注

配置

使用情况

事件元数据

连接

host.json 属性

后续步骤

其他资源