c# - 从哪里获取 Azure Durable Functions 中 RaiseEventAsync 方法的 InstanceId？

Question

我正在学习如何使用 Azure Durable Functions，但遇到了问题。我正在向队列发布一条消息并等待它，然后将其记录下来进行跟踪，但我不确定从哪里获取 InstanceId。

这是我到目前为止所拥有的:

public static class Go
{
    private const string ReceivedMessage = "received-message";

    [FunctionName(nameof(Go))]
    public static async Task Run(
        [OrchestrationTrigger] DurableOrchestrationContext context,
        TraceWriter log)
    {
        var message = await context.WaitForExternalEvent<string>(ReceivedMessage);

        log.Info($"Received message: {message}");
    }

    [FunctionName(nameof(GetMessageFromQueue))]
    public static async Task GetMessageFromQueue(
        [QueueTrigger("messages")] string message,
        [OrchestrationClient] DurableOrchestrationClient client)
    {
        await client.RaiseEventAsync("InstanceId", ReceivedMessage, message); // <- here
    }
}

为了完整起见，这是我拥有的其余代码。 HttpStartSingle 类与上面的项目位于同一项目中，Program 类只是一个普通的控制台应用程序，我用它来启动上下文:

public static class HttpStartSingle
{
    [FunctionName(nameof(HttpStartSingle))]
    public static async Task<HttpResponseMessage> RunSingle(
        [HttpTrigger(
            AuthorizationLevel.Function,
            "get", "post",
            Route = "orchestrators/{functionName}")]
        HttpRequestMessage req,
        [OrchestrationClient] DurableOrchestrationClient starter,
        string functionName,
        TraceWriter log)
    {
        var eventData = await req.Content.ReadAsAsync<object>();
        var instanceId = await starter.StartNewAsync(functionName, eventData);

        return starter.CreateCheckStatusResponse(req, instanceId);
    }
}

public class Program
{
    public static async Task Main(string[] args)
    {
        Thread.Sleep(TimeSpan.FromSeconds(5));

        var request = WebRequest.Create("http://localhost:7071/api/orchestrators/Go");

        request.Timeout = Convert.ToInt32(TimeSpan.FromMinutes(1).TotalSeconds);
        request.ContentLength = 0;
        request.Method = "POST";

        var json = string.Empty;

        using (var response = await request.GetResponseAsync())
        using (var stream = response.GetResponseStream())
        {
            if (stream != null)
            {
                using (var reader = new StreamReader(stream, Encoding.UTF8))
                {
                    if (reader.Peek() > -1)
                    {
                        json = await reader.ReadToEndAsync();
                    }
                }
            }
        }

        Console.WriteLine(json);

        var storageAccount = CloudStorageAccount.DevelopmentStorageAccount;
        var queueuClient = storageAccount.CreateCloudQueueClient();
        var queue = queueuClient.GetQueueReference("messages");

        await queue.CreateIfNotExistsAsync();
        await queue.AddMessageAsync(new CloudQueueMessage("This is a test message"));

        Console.ReadKey();
    }
}

更新:

上面的代码实际上只是一个概念证明，看看 Durable Functions 是否满足我的要求。到目前为止我还不确定，但这就是我想要做的:

我将有两个队列。

队列 A 将使用少量 A 作为种子，我打算回收这些 A。

队列 B 将有一个由外部源推送到它的不确定数量的 B(对于所有意图和目的而言都是无限的)。

我希望协调器等待来自两个队列的输入，然后将 A 和 B 发送到另一个函数，该函数将生成多个 B。然后，该函数会将 B 推送到队列 B，然后将 A 推送回队列 A。然后重复该过程。

我希望根据我拥有的 A 数量并行处理。在我看来，我有两个选择:

A) 创建一个使用取自常量的 InstanceId 创建的 Orchestrator。然后，监听 A 和 B 队列的函数将知道用于 RaiseEventAsync 调用的 InstanceId。

问题:不并行。

B) 在 A 中包装一个实例 ID。为每个 A 创建一个 Orchestrator。A 知道要使用哪个 Orchestrator，现在每个 Orchestrator 只能监听队列 B。

问题:B 不知道 RaiseEventAsync 调用要使用什么 InstanceId。

Answer 1

无论将消息放入队列中，将触发您的 GetMessageFromQueue 都需要了解它尝试定位的实例，然后我希望您在消息本身中包含实例 ID。

在您上面列出的特定场景中，我实际上希望基于 HttpTrigger 的函数返回包含实例 ID 的正文。现在，您正在返回由 CreateCheckStatusResponse 创建的 HttpResponseMessage，但这被记录为返回带有 Location< 的 202 状态 header 告诉人们可以在哪里检查该编排实例的状态。虽然此信息对于交还可能也很有用，但您确实需要交还实例 ID。

一种方法是创建您自己的 HttpResponseMessage，其中包含强类型消息正文，其中包含实例 ID 作为属性。如果您这样做，那么调用程序将能够从响应正文中提取该 ID，然后将其包含在放入队列的消息中，以便您的 GetMessageFromQueue 函数可以将其从在那里传递给 RaiseEventAsync。