水平扩展写入时如何避免并发问题？

Question

假设有一个工作服务从队列接收消息，从文档数据库中读取具有指定 Id 的产品，根据消息应用一些操作逻辑，最后将更新后的产品写回数据库 (a) .

当处理不同的产品时，这项工作可以安全地并行完成，因此我们可以水平扩展 (b)。但是，如果多个服务实例在同一个产品上工作，我们最终可能会遇到并发问题，或者数据库中的并发异常，在这种情况下，我们应该应用一些重试逻辑（但重试仍然可能再次失败等等） .

问题：我们如何避免这种情况？有没有办法确保两个实例不在同一个产品上运行？

示例/用例：一家在线商店在产品 A、产品 B 和产品 C 上进行了一次大促销，在一个小时内结束，数百名客户正在购买。对于每次购买，都会将一条消息排入队列（productId、numberOfItems、price）。 目标：我们如何运行我们的工作服务的三个实例，并确保 productA 的所有消息都将在 instanceA、productB 到 instanceB 和 productC 到 instanceC 中结束（不会导致并发问题）？

注意事项：我的服务是用 C# 编写的，作为辅助角色托管在 Azure 上，我使用 Azure Queues 进行消息传递，并且我正在考虑使用 Mongo 进行存储。此外，实体 ID 为 GUID。

更多的是关于技术/设计，所以如果您使用不同的工具来解决问题，我仍然感兴趣。

Answer 1

我假设您有办法在所有工作人员服务中安全地访问产品队列。鉴于此，避免冲突的一种简单方法是在主队列旁边为每个产品使用全局队列

// Queue[X] is the queue for product X
// QueueMain is the main queue 
DoWork(ProductType X)
{
  if (Queue[X].empty())
  {
    product = QueueMain().pop()
    if (product.type != X)
    {
      Queue[product.type].push(product) 
      return;
    }
  }else
  {
     product = Queue[X].pop()
  }

  //process product...
}

对队列的访问需要是原子的

Answer 2

1) 我能想到的每一个大规模数据解决方案都内置了一些东西来精确处理这种冲突。详细信息将取决于您对数据存储的最终选择。对于传统的关系数据库，无需您进行任何附加工作即可完成此操作。有关详细信息，请参阅您选择的技术文档。

2) 了解您的数据模型和使用模式。适当地设计您的数据存储。不要为你没有的规模设计。针对最常见的使用模式进行优化。

3) 挑战你的假设。你真的不得不频繁地从多个角色中改变同一个实体吗？有时答案是肯定的，但通常您可以简单地创建一个类似的新实体来反映更新。 IE，采用 journaling/logging 方法而不是单一实体方法。最终，单个实体的大量更新永远不会扩展。

Answer 3

如果您希望数据库始终保持最新并且始终与已处理的单元保持一致，那么您可以对同一个可变实体进行多次更新。

为了遵守这一点，您需要序列化同一实体的更新。要么通过在生产者处对数据进行分区来做到这一点，要么将实体的事件累积在同一队列中，要么使用分布式锁或数据库级别的锁将实体锁定在工作线程中。

您可以使用 Actor 模型（在使用 akka 的 java/scala 世界中）为每个实体或串行处理它们的实体组创建一个消息队列。

更新 你可以试试akka port to .net 和here。 在这里，您可以找到一个很好的教程，其中包含有关使用 akka in scala 的示例。 但是对于一般原则，您应该更多地搜索 [actor model]。但它也有缺点。

最终涉及对您的数据进行分区以及为特定实体创建独特的专业工作者（可以重复使用和/或在发生故障时重新启动）的能力。

Answer 4

任何试图将负载分配到同一集合中的不同项目（如订单）的解决方案都注定要失败。原因是，如果您的交易流量很高，则必须开始执行以下操作之一：

1. 让节点相互交谈 (hey guys, are anyone working with this?)
2. 将ID生成划分为段（节点a创建ID 1-1000，节点B 1001-1999）等，然后让它们处理自己的段
3. 动态地将集合划分为段（并让每个节点处理一个段。

那么这些方法有什么问题？

第一种方法是简单地复制数据库中的事务。除非您可以花费大量时间优化策​​略，否则最好依赖交易。

后两个选项会降低性能，因为您必须根据 id 动态路由消息，并且还要在运行时更改策略以包括新插入的消息。它最终会失败。

解决方案

这里有两种解决方案，您也可以组合使用。

自动重试

相反，您在某处有一个从消息队列读取的入口点。

你有这样的东西：

while (true)
{
    var message = queue.Read();
    Process(message);
}

为了获得非常简单的容错能力，您可以做的是在失败时重试：

while (true)
{
    for (i = 0; i < 3; i++)
    {
       try
       {
            var message = queue.Read();
            Process(message);
            break; //exit for loop
       }
       catch (Exception ex)
       {
           //log
           //no throw = for loop runs the next attempt
       }
    }
}

您当然可以只捕获数据库异常（或者更确切地说是事务失败）来重播这些消息。

微服务

我知道，微服务是一个流行词。但在这种情况下，这是一个很好的解决方案。与其拥有一个处理所有消息的单一核心，不如将应用程序分成更小的部分。或者在您的情况下，只需停用某些类型的消息的处理。

如果您有五个节点运行您的应用程序，您可以确保节点 A 接收与订单相关的消息，节点 B 接收与运输等相关的消息。

通过这样做，您仍然可以水平扩展您的应用程序，您不会遇到任何冲突并且需要很少的努力（更多的消息队列并重新配置每个节点）。

Answer 5

对于这种事情，我使用 blob 租约。基本上，我使用某个已知存储帐户中的实体 ID 创建了一个 blob。当 worker 1 拿起实体时，它会尝试获取 blob 的租约（如果 blob 不存在，则自行创建 blob）。如果两者都成功，那么我允许对消息进行处理。之后总是释放租约。 如果我不成功，我将消息转储回队列中

我遵循 Steve Marx 最初在此处 http://blog.smarx.com/posts/managing-concurrency-in-windows-azure-with-leases 描述的方法，尽管经过调整以使用新的存储库

在 cmets 之后编辑： 如果您可能有很高的消息都与同一个实体交谈（正如您的推荐所暗示的那样），我会在某处重新设计您的方法......无论是实体结构还是消息结构。

例如：考虑 CQRS 设计模式并独立存储来自处理每条消息的更改。因此，产品实体现在是各种工作人员对实体所做的所有更改的集合，依次重新应用并重新水化为单个对象