你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。
适用于 Azure Functions 的 Azure 事件中心触发器
本文介绍如何使用适用于 Azure Functions 的 Azure 事件中心触发器。 Azure Functions 支持事件中心的触发器和输出绑定。
有关设置和配置详细信息，请参阅概述。
使用函数触发器来响应发送到事件中心事件流的事件。 若要设置触发器，必须具有基础事件中心的读取访问权限。 触发函数时，传递给函数的消息充当字符串类型。
面向消耗计划和高级计划的事件中心缩放决策是通过基于目标的缩放完成的。 有关详细信息，请参阅基于目标的缩放。
有关 Azure Functions 如何使用触发器响应发送到事件中心事件流的事件的信息，请参阅将事件中心与 Azure 上的无服务器函数集成。
重要
本文使用选项卡来支持多个版本的 Node.js 编程模型。 v4 模型已正式发布，旨在为 JavaScript 和 TypeScript 开发人员提供更为灵活和直观的体验。 有关 v4 模型工作原理的更多详细信息，请参阅 Azure Functions Node.js 开发人员指南。 要详细了解 v3 和 v4 之间的差异，请参阅迁移指南。
Azure Functions 支持两种 Python 编程模型。 定义绑定的方式取决于选择的编程模型。
使用 Python v2 编程模型，可以直接在 Python 函数代码中使用修饰器定义绑定。 有关详细信息，请参阅 Python 开发人员指南。
本文同时支持两个编程模型。
示例
以下示例演示了一个 C# 函数，该函数基于事件中心而触发，其中输入消息字符串会写入到日志中：
{
private readonly ILogger<EventHubsFunction> _logger;
public EventHubsFunction(ILogger<EventHubsFunction> logger)
{
_logger = logger;
}
[Function(nameof(EventHubFunction))]
[FixedDelayRetry(5, "00:00:10")]
[EventHubOutput("dest", Connection = "EventHubConnection")]
public string EventHubFunction(
[EventHubTrigger("src", Connection = "EventHubConnection")] string[] input,
FunctionContext context)
{
_logger.LogInformation("First Event Hubs triggered message: {msg}", input[0]);
var message = $"Output message created at {DateTime.Now}";
return message;
}
以下示例展示了事件中心触发器 - TypeScript 函数。 此函数将读取事件元数据并记录消息。
import { app, InvocationContext } from '@azure/functions';
export async function eventHubTrigger1(message: unknown, context: InvocationContext): Promise<void> {
context.log('Event hub function processed message:', message);
context.log('EnqueuedTimeUtc =', context.triggerMetadata.enqueuedTimeUtc);
context.log('SequenceNumber =', context.triggerMetadata.sequenceNumber);
context.log('Offset =', context.triggerMetadata.offset);
}
app.eventHub('eventHubTrigger1', {
connection: 'myEventHubReadConnectionAppSetting',
eventHubName: 'MyEventHub',
cardinality: 'one',
handler: eventHubTrigger1,
});
若要批量接收事件，请将 cardinality 设置为 many，如以下示例所示。
import { app, InvocationContext } from '@azure/functions';
export async function eventHubTrigger1(messages: unknown[], context: InvocationContext): Promise<void> {
context.log(`Event hub function processed ${messages.length} messages`);
for (let i = 0; i < messages.length; i++) {
context.log('Event hub message:', messages[i]);
context.log(`EnqueuedTimeUtc = ${context.triggerMetadata.enqueuedTimeUtcArray[i]}`);
context.log(`SequenceNumber = ${context.triggerMetadata.sequenceNumberArray[i]}`);
context.log(`Offset = ${context.triggerMetadata.offsetArray[i]}`);
}
}
app.eventHub('eventHubTrigger1', {
connection: 'myEventHubReadConnectionAppSetting',
eventHubName: 'MyEventHub',
cardinality: 'many',
handler: eventHubTrigger1,
});
以下示例展示了事件中心触发器 - JavaScript 函数。 此函数将读取事件元数据并记录消息。
const { app } = require('@azure/functions');
app.eventHub('eventHubTrigger1', {
connection: 'myEventHubReadConnectionAppSetting',
eventHubName: 'MyEventHub',
cardinality: 'one',
handler: (message, context) => {
context.log('Event hub function processed message:', message);
context.log('EnqueuedTimeUtc =', context.triggerMetadata.enqueuedTimeUtc);
context.log('SequenceNumber =', context.triggerMetadata.sequenceNumber);
context.log('Offset =', context.triggerMetadata.offset);
},
});
若要批量接收事件，请将 cardinality 设置为 many，如以下示例所示。
const { app } = require('@azure/functions');
app.eventHub('eventHubTrigger1', {
connection: 'myEventHubReadConnectionAppSetting',
eventHubName: 'MyEventHub',
cardinality: 'many',
handler: (messages, context) => {
context.log(`Event hub function processed ${messages.length} messages`);
for (let i = 0; i < messages.length; i++) {
context.log('Event hub message:', messages[i]);
context.log(`EnqueuedTimeUtc = ${context.triggerMetadata.enqueuedTimeUtcArray[i]}`);
context.log(`SequenceNumber = ${context.triggerMetadata.sequenceNumberArray[i]}`);
context.log(`Offset = ${context.triggerMetadata.offsetArray[i]}`);
}
},
});
下面是 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'))
以下示例演示了一个记录事件中心触发器消息正文的事件中心触发器绑定。
@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);
}
}
修饰符
仅适用于 Python v2 编程模型。
对于使用修饰器定义的 Python v2 功能，支持 event_hub_message_trigger 上的以下属性：
| properties | 说明 |
|---|---|
arg_name |
在函数代码中表示事件项的变量的名称。 |
event_hub_name |
事件中心的名称。 当事件中心名称也出现在连接字符串中时，该值会在运行时覆盖此属性。 |
connection |
指定如何连接到事件中心的应用设置或设置集合的名称。 请参阅连接。 |
对于使用 function.json 定义的 Python 函数，请参阅“配置”部分。
配置
仅适用于 Python v1 编程模型。
下表说明了在 function.json 文件中设置的触发器配置属性，这些属性因运行时版本而异。
| function.json 属性 | 说明 |
|---|---|
| type | 必须设置为 eventHubTrigger。 在 Azure 门户中创建触发器时，会自动设置此属性。 |
| direction | 必须设置为 in。 在 Azure 门户中创建触发器时，会自动设置此属性。 |
| name | 在函数代码中表示事件项的变量的名称。 |
| eventHubName | 事件中心的名称。 当事件中心名称也出现在连接字符串中时，该值会在运行时覆盖此属性。 可以通过应用设置 %eventHubName% 进行引用 |
| consumerGroup | 一个可选属性，用于设置使用者组，该组用于订阅事件中心中的事件。 如果将其省略，则会使用 $Default 使用者组。 |
| 基数 | 设为 many 以启用批处理。 如果省略或设为 one，将向函数传递一条消息。 |
| 连接 | 指定如何连接到事件中心的应用设置或设置集合的名称。 请参阅连接。 |
在本地开发时，需要将应用程序设置添加到 Values 集合中的 local.settings.json 文件中。
使用情况
要详细了解事件中心触发器和 IoT 中心触发器的缩放方式，请参阅通过 Azure Functions 使用事件。
事件中心输出绑定支持的参数类型取决于所用的 Functions 运行时版本、扩展包版本以及 C# 模态。
如果你希望函数处理单个事件，可将事件中心触发器绑定到以下类型：
| 类型 | 说明 |
|---|---|
string |
字符串形式的事件。 当事件为简单文本时使用。 |
byte[] |
事件的字节数。 |
| JSON 可序列化类型 | 当事件包含 JSON 数据时，Functions 会尝试将 JSON 数据反序列化为普通的旧 CLR 对象 (POCO) 类型。 |
| Azure.Messaging.EventHubs.EventData1 | 事件对象。 如果要从事件中心 SDK 的任何旧版本迁移，请注意，此版本将放弃对旧 Body 类型的支持，转而支持 EventBody。 |
如果你希望函数处理批量事件，可将事件中心触发器绑定到以下类型：
| 类型 | 说明 |
|---|---|
string[] |
批处理中的事件数组，以字符串形式表示。 每个条目表示一个事件。 |
EventData[] 1 |
批处理中的事件数组，以 Azure.Messaging.EventHubs.EventData 的实例形式表示。 每个条目表示一个事件。 |
T[]，其中 T 是 JSON 可序列化类型1 |
批处理中的事件数组，以自定义 POCO 类型的实例形式表示。 每个条目表示一个事件。 |
1 若要使用这些类型，需要引用 Microsoft.Azure.Functions.Worker.Extensions.EventHubs 5.5.0 或更高版本以及 SDK 类型绑定的常见依赖项。
参数类型可以为以下类型之一：
- 任意的本机 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 属性指定的值匹配。
基于标识的连接
如果使用 5.x 或更高版本的扩展，可以让应用使用 Microsoft Entra 标识，而无需将连接字符串与机密一起使用。 为此，需要定义公共前缀下的设置，该前缀映射到触发器和绑定配置中的 connection 属性。
在此模式下，扩展需要以下属性：
| 属性 | 环境变量模板 | 说明 | 示例值 |
|---|---|---|---|
| 完全限定的命名空间 | <CONNECTION_NAME_PREFIX>__fullyQualifiedNamespace |
完全限定的事件中心命名空间。 | myeventhubns.servicebus.windows.net |
可以设置其他属性来自定义连接。 请参阅基于标识的连接的通用属性。
注意
使用 Azure 应用程序配置或 Key Vault 为托管标识连接提供设置时，设置名称应使用有效的键分隔符（例如 : 或 /）替代 __，以确保正确解析名称。
例如 <CONNECTION_NAME_PREFIX>:fullyQualifiedNamespace。
在 Azure Functions 服务中托管时，基于标识的连接将使用托管标识。 默认情况下使用系统分配的标识，但可以使用 credential 和 clientID 属性来指定用户分配的标识。 请注意，不支持为用户分配的标识配置资源 ID。 在其他上下文（如本地开发）中运行时，将改用开发人员标识，尽管可以进行自定义。 请参阅使用基于标识的连接进行本地开发。
向标识授予权限
无论使用何种标识，都必须具有执行所需操作的权限。 对于大多数 Azure 服务，这意味着你需要使用内置角色或者提供这些权限的自定义角色在 Azure RBAC 中分配角色。
重要
某些权限可能由并非所有上下文都需要的目标服务公开。 尽可能遵循最低权限原则，仅授予标识所需的权限。 例如，如果应用只需要从数据源进行读取即可，则使用仅具有读取权限的角色。 分配一个也具有该服务写入权限的角色并不恰当，因为对于读取操作来说，写入是多余的权限。 同样，你也希望确保角色分配的范围仅限于需要读取的资源。
你将需要创建一个角色分配，以便在运行时提供对事件中心的访问权限。 角色分配的范围可以是事件中心命名空间，也可以是事件中心本身。 所有者等管理角色还不够。 下表显示了在正常操作中使用事件中心扩展时建议使用的内置角色。 根据所编写的代码，应用程序可能需要具有其他权限。
| 绑定类型 | 内置角色示例 |
|---|---|
| 触发器 | Azure 事件中心数据接收方、Azure 事件中心数据所有者 |
| 输出绑定 | Azure 事件中心数据发送方 |
host.json 设置
host.json 文件包含控制事件中心触发器行为的设置。 有关可用设置的详细信息，请参阅 host.json 设置部分。