适用于 Azure Functions 的 Azure 事件中心输出绑定

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

Python v1 编程模型要求在函数文件夹中的单独 function.json 文件中定义绑定。 有关详细信息，请参阅 Python 开发人员指南。

以下示例显示了使用方法返回值作为输出将消息字符串写入到事件中心的 C# 函数：

        [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;
        }

以下示例演示使用方法返回值作为输出将消息写入到事件中心的 C# 函数：

[FunctionName("EventHubOutput")]
[return: EventHub("outputEventHubMessage", Connection = "EventHubConnectionAppSetting")]
public static string Run([TimerTrigger("0 */5 * * * *")] TimerInfo myTimer, ILogger log)
{
    log.LogInformation($"C# Timer trigger function executed at: {DateTime.Now}");
    return $"{DateTime.Now}";
}

以下示例演示如何使用 IAsyncCollector 接口发送一批消息。 在处理来自一个事件中心的消息并将结果发送到另一个事件中心时，这种情况很常见。

[FunctionName("EH2EH")]
public static async Task Run(
    [EventHubTrigger("source", Connection = "EventHubConnectionAppSetting")] EventData[] events,
    [EventHub("dest", Connection = "EventHubConnectionAppSetting")]IAsyncCollector<EventData> outputEvents,
    ILogger log)
{
    foreach (EventData eventData in events)
    {
        // Do some processing:
        string newEventBody = DoSomething(eventData);

        // Queue the message to be sent in the background by adding it to the collector.
        // If only the event is passed, an Event Hubs partition to be assigned via
        // round-robin for each batch.
        await outputEvents.AddAsync(new EventData(newEventBody));

        // If your scenario requires that certain events are grouped together in an
        // Event Hubs partition, you can specify a partition key.  Events added with 
        // the same key will always be assigned to the same partition.        
        await outputEvents.AddAsync(new EventData(newEventBody), "sample-key");
    }
}

以下示例显示计时器触发的 TypeScript 函数，该函数将一条消息发送到事件中心：

import { app, InvocationContext, output, Timer } from '@azure/functions';

export async function timerTrigger1(myTimer: Timer, context: InvocationContext): Promise<string> {
    const timeStamp = new Date().toISOString();
    return `Message created at: ${timeStamp}`;
}

app.timer('timerTrigger1', {
    schedule: '0 */5 * * * *',
    return: output.eventHub({
        eventHubName: 'myeventhub',
        connection: 'MyEventHubSendAppSetting',
    }),
    handler: timerTrigger1,
});

若要输出多条消息，请返回一个数组而不是单个对象。 例如：

import { app, InvocationContext, output, Timer } from '@azure/functions';

export async function timerTrigger1(myTimer: Timer, context: InvocationContext): Promise<string[]> {
    // <displayInDocs>
    const timeStamp = new Date().toISOString();
    const message = `Message created at: ${timeStamp}`;
    return [`1: ${message}`, `2: ${message}`];
    // </displayInDocs>
}

app.timer('timerTrigger1', {
    schedule: '0 */5 * * * *',
    return: output.eventHub({
        eventHubName: 'myeventhub',
        connection: 'MyEventHubSendAppSetting',
    }),
    handler: timerTrigger1,
});

模型 v3 没有记录 TypeScript 示例。

以下示例显示计时器触发的 JavaScript 函数，该函数将一条消息发送到事件中心：

const { app, output } = require('@azure/functions');

const eventHubOutput = output.eventHub({
    eventHubName: 'myeventhub',
    connection: 'MyEventHubSendAppSetting',
});

app.timer('timerTrigger1', {
    schedule: '0 */5 * * * *',
    return: eventHubOutput,
    handler: (myTimer, context) => {
        const timeStamp = new Date().toISOString();
        return `Message created at: ${timeStamp}`;
    },
});

若要输出多条消息，请返回一个数组而不是单个对象。 例如：

const { app, output } = require('@azure/functions');

const eventHubOutput = output.eventHub({
    eventHubName: 'myeventhub',
    connection: 'MyEventHubSendAppSetting',
});

app.timer('timerTrigger1', {
    schedule: '0 */5 * * * *',
    return: eventHubOutput,
    handler: (myTimer, context) => {
        // <displayInDocs>
        const timeStamp = new Date().toISOString();
        const message = `Message created at: ${timeStamp}`;
        return [`1: ${message}`, `2: ${message}`];
        // </displayInDocs>
    },
});

以下示例显示了 function.json 文件中的一个事件中心触发器绑定以及使用该绑定的函数。 该函数将输出消息写入事件中心。

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

{
    "type": "eventHub",
    "name": "outputEventHubMessage",
    "eventHubName": "myeventhub",
    "connection": "MyEventHubSendAppSetting",
    "direction": "out"
}

下面是可发送一条消息的 JavaScript 代码：

module.exports = function (context, myTimer) {
    var timeStamp = new Date().toISOString();
    context.log('Message created at: ', timeStamp);   
    context.bindings.outputEventHubMessage = "Message created at: " + timeStamp;
    context.done();
};

下面是可发送多条消息的 JavaScript 代码：

module.exports = function(context) {
    var timeStamp = new Date().toISOString();
    var message = 'Message created at: ' + timeStamp;

    context.bindings.outputEventHubMessage = [];

    context.bindings.outputEventHubMessage.push("1 " + message);
    context.bindings.outputEventHubMessage.push("2 " + message);
    context.done();
};

import logging
import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="eventhub_output")
@app.route(route="eventhub_output")
@app.event_hub_output(arg_name="event",
                      event_hub_name="<EVENT_HUB_NAME>",
                      connection="<CONNECTION_SETTING>")
def eventhub_output(req: func.HttpRequest, event: func.Out[str]):
    body = req.get_body()
    if body is not None:
        event.set(body.decode('utf-8'))
    else:    
        logging.info('req body is none')
    return 'ok'

下面是可发送多条消息的 Python 代码：

import logging
import azure.functions as func
from typing import List

app = func.FunctionApp()

@app.function_name(name="eventhub_output")
@app.route(route="eventhub_output")
@app.event_hub_output(arg_name="event",
                      event_hub_name="<EVENT_HUB_NAME>",
                      connection="<CONNECTION_SETTING>")

def eventhub_output(req: func.HttpRequest, event: func.Out[List[str]]) -> func.HttpResponse:
    my_messages=["message1", "message2","message3"]
    event.set(my_messages)
    return func.HttpResponse(f"Messages sent")

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

{
    "type": "eventHub",
    "name": "$return",
    "eventHubName": "myeventhub",
    "connection": "MyEventHubSendAppSetting",
    "direction": "out"
}

下面是可发送一条消息的 Python 代码：

import datetime
import logging
import azure.functions as func


def main(timer: func.TimerRequest) -> str:
    timestamp = datetime.datetime.utcnow()
    logging.info('Message created at: %s', timestamp)
    return 'Message created at: {}'.format(timestamp)

下面是可发送多条消息的 Python 代码：

import logging
from typing import List
import azure.functions as func


def main(req: func.HttpRequest, messages:func.Out[List[str]]) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    messages.set([{"message1"}, {"message2"}])
    return func.HttpResponse(f"Messages sent")

使用 [EventHubOutputAttribute] 定义到事件中心的输出绑定，EventHubOutputAttribute 支持以下属性。

使用 EventHubAttribute 定义到事件中心的输出绑定，EventHubAttribute 支持以下属性。

下表说明了可以在传递给 output.eventHub() 方法的 options 对象上设置的属性。

下表解释了在 function.json 文件中设置的绑定配置属性。

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

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

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

可以使用 out string paramName 等方法参数发送消息。 若要编写多条消息，可以使用 ICollector<EventData> 或 IAsyncCollector<EventData> 代替 out string。 分区键只能与 IAsyncCollector<EventData> 一起使用。

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

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

可以使用 out string paramName 等方法参数发送消息。 若要编写多条消息，可以使用 ICollector<string> 或 IAsyncCollector<string> 代替 out string。

如果希望函数写入单个事件，事件中心输出绑定可以绑定到以下类型：

如果希望函数写入多个事件，事件中心输出绑定可以绑定到以下类型：

对于其他输出方案，请直接从 Microsoft.Azure.EventHubs 创建和使用类型。

需要定义自定义类型或使用字符串。 扩展 v5.x+ 中提供了其他选项。

通过直接返回值或使用 context.extraOutputs.set() 来访问输出消息。

使用 context.bindings.<name> 访问输出事件，其中 <name> 是 function.json 的 name 属性中指定的值。