Azure 无服务器上的可扩展有序消息处理答案

【问题标题】：Scalable in-order message processing on Azure serverlessAzure 无服务器上的可扩展有序消息处理
【发布时间】：2018-12-14 14:13:04
【问题描述】：

我需要在 Azure 上创建一些可以处理一组实体的传入消息流的东西。我们将在任何时间点拥有 20 到 2,000 个实体；这些被动态创建和丢弃。消息将使用我们的本地系统生成，并使用某种排队机制发送到 Azure。每条消息都将通过EntityId 属性与特定实体相关联。属于同一实体的消息必须相互按顺序进行处理。

同时，解决方案必须在实体方面具有可扩展性。如果我有 1,000 个实体的稳定消息流，我希望我的逻辑有 1,000 次并发执行。如果一个实体需要很长时间来处理它的一个消息，这不能阻止任何其他实体处理 他们的 消息。每条消息可能需要 100 毫秒到 10 秒的时间来处理（绝大多数低于 1 秒），每个实体平均每秒会收到一条消息。

令人失望的是，Azure 无服务器堆栈似乎没有任何方法可以实现这一点。这些是我考虑过的选项及其问题：

由带有会话的服务总线队列触发的 Azure Functions。 Azure Functions 可以在消费计划中作为无服务器运行，使其非常适合弹性扩展。服务总线会话提供按顺序交付，并且是最接近我的要求的实现。但是，它们在 Azure Functions 中不受支持：Support Service Bus queues and topics which use sessions。
由带有会话的服务总线队列触发的逻辑应用：通过“Correlated in-order delivery using service bus sessions”模板开箱即用地支持这一点。然后，逻辑应用可以连接到 HTTP 触发的 Azure 函数以处理消息。逻辑应用的唯一目的是防止同时处理属于同一实体/会话的多条消息。但是，从 cmets 到我的 former question，我发现这也无法扩展。一个逻辑应用每 5 分钟只能执行 300,000 个动作，触发并发限制为 50，据说价格昂贵。见Limits and configuration information for Azure Logic Apps。
带有分区的 Azure 事件中心，如 In order event processing with Azure Functions 中所述。这往往是最受欢迎的选项，也是 Microsoft 推荐的选项。但是，事件中心最多只允许 32 个分区，需要在创建时指定分区数：Features and terminology in Azure Event Hubs。这种固定静态分区集的限制违背了“无服务器”的精神；如果我们将并行度限制为 32，那么我们不会获得比在 32 核机器上运行的并行应用程序更好的可扩展性。可以通过向 Microsoft 提供支持票证将分区限制增加到 32 个以上，但我不想要求可扩展性超出一般用途的两个数量级。事件中心还缺少一些其他基本属性，例如最多一次交付。
我们可以为每个实体动态创建一个服务总线队列，并为其生成一个单独的 Azure 函数，专门绑定到该特定队列。但是，这需要调用 Azure 资源管理 API 作为我们操作代码的一部分，而我的印象是 Azure Functions 并非旨在以这种方式动态生成。
使用服务总线队列消息的SequenceNumber 属性对持久性后备存储（如 Redis）进行优化并发控制以进行排序。然而，用于此的编程模型非常复杂且容易出错——操作需要在重试循环中执行，并且每次都明确考虑原子性和幂等性。此外，它要求所有消息都包含完整的实体状态；否则，当我们在比赛中丢弃陈旧的消息时，信息将会丢失。对于我们来说（在计算和带宽方面）随着每次增量更改发送完整的实体快照会很昂贵，因此我们宁愿找到一种方法来允许我们按顺序处理增量消息。

是否有任何干净的方法可以在 Azure 无服务器堆栈上为可扩展数量的实体实现有序处理？

【问题讨论】：

看起来 Event Hubs Dedicated 集群可以为您的解决方案提供帮助，docs.microsoft.com/en-us/azure/event-hubs/…，您的专用集群可以根据您的需求进行配置，例如 TU 数量、分区等。这是自定义Microsoft 根据您的要求完成的单租户 EH 部署。
感谢@RomanKiss，这看起来是唯一的解决方案。我需要调查一下它的成本。

标签： azure azure-functions azure-logic-apps serverless

【解决方案1】：

Azure Functions 现在使用 Microsoft Azure WebJobs SDK ServiceBus Extension (Microsoft.Azure.WebJobs.Extensions.ServiceBus) 3.1.0 或更高版本支持服务总线会话。例如：

public async Task Run(
    [ServiceBusTrigger("queue", Connection = "ServiceBusConnectionString", IsSessionsEnabled = true)]Message message, 
    ILogger log)
{
    log.LogInformation($"C# ServiceBus queue trigger function processed message: {Encoding.UTF8.GetString(message.Body)}");
    await _client.PushData(message.SessionId, Encoding.UTF8.GetString(message.Body));
}

见Ordered queue processing in Azure Functions with Sessions。

【讨论】：