Azure EventHub 和持久函数

Question

实际上是在尝试做一些我不擅长的事情。

我在这里阅读了持久功能概述 - https://learn.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-overview 。

有一个关于使用绑定(bind)在事件中心触发器上使用它的主题，但实际上并没有一个有效的示例，我按照那里的内容并在我的 function.json,

{
  "bindings": [
    {
      "type": "eventHubTrigger",
      "name": "myEventHubMessage",
      "direction": "in",
      "path": "testinhub",
      "connection": "Endpoint=sb://dev-testingeventhubinns.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=lLassdff5Y/esH8/CaXDOWH0jF2JtZBQhQeFoCtfqYs=",
      "consumerGroup": "$Default"
    },
    {
      "type": "eventHub",
      "name": "outputEventHubMessage",
      "connection": "Endpoint=sb://dev-testingeventhuboutns.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=4yuasdff7Lzu+mQJFVlnlozUItqFY1L3WW/kJnpTjq8=",
      "path": "testouthub",
      "direction": "out"
    }
  ],
  "disabled": false,
  "entryPoint": "EventTriggerFunction.EventHubTriggerClass.Run"
}

我的完整代码如下，

using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using Microsoft.Azure.EventHubs;
using System;

namespace EventTriggerFunction
{
    public static class EventHubTriggerClass
    {
        [FunctionName("EventHubTrigger")]
        public static async Task<List<string>> Run([OrchestrationTrigger] DurableOrchestrationContext context)
        {
            await context.CallActivityAsync<string>("EventHubTrigger_Send", "Hello World");

            return null;
        }

        [FunctionName("EventHubTrigger_Send")]
        public static void SendMessages([EventHubTrigger("testinhub", Connection = "ConnectionValue")] EventData[] eventHubMessages, ILogger log)
        {
            var exceptions = new List<Exception>();

            foreach (EventData message in eventHubMessages)
            {
                try
                {
                    log.LogInformation($"C# Event Hub trigger function processed a message: {Encoding.UTF8.GetString(message.Body)}");
                }
                catch (Exception e)
                {
                    // We need to keep processing the rest of the batch - capture this exception and continue.
                    // Also, consider capturing details of the message that failed processing so it can be processed again later.
                    exceptions.Add(e);
                }
            }            
        }
    }
}

如果我使用该函数发送消息，我将无法在我的 testouthub 事件中心上看到它。不太确定这个 Durable 函数和 EventHub 触发器如何协同工作。

Answer 1

我认为你有点混淆了。使用 FunctionName 和 EventHubTrigger 等属性时，不需要提供 function.json。它可以是 function.json 或带有属性。

如果您尝试从 1 个 eventhub 接收消息并将其传递到下一个 eventhub，那么下面类似的操作也可以解决问题。无需为此使用 DurableFunctions，如果消息较多，Azure Function 运行时将自行扩展，请参阅 this

[FunctionName("EventHubTriggerCSharp")]
[return: EventHub("outputEventHubMessage", Connection = "EventHubConnectionAppSetting")]
public static void Run([EventHubTrigger("samples-workitems", Connection = "EventHubConnectionAppSetting")] string myEventHubMessage, ILogger log)
{
    log.LogInformation($"C# Event Hub trigger function processed a message: {myEventHubMessage}");

    return myEventHubMessage;
}

额外提示:我不会将您的完整连接字符串粘贴到 StackOverflow 中。也许立即为您的事件中心创建新的 AccessKey 是明智的。