使用者竞争模式

使多个并发使用者能够处理同一消息通道上收到的消息。存在多个并发使用者时，系统可以并发处理多个消息，以优化吞吐量、改进可伸缩性和可用性，以及均衡工作负载。

上下文和问题

在云中运行的应用程序需要处理大量的请求。常用方法不是同步处理每个请求，而是应用程序通过消息传递系统将它们传送到异步处理它们的另一个服务（使用者服务）。此策略有助于确保在处理请求时应用程序中的业务逻辑不会被阻止。在一段时间内，由于多种原因请求的数量会大幅度变化。用户活动或来自多个租户的总请求数的突增可能会导致不可预测的工作负荷。在高峰期，系统可能需要每秒处理数百个请求，而在其他时间，请求的数量可能非常少。此外，为处理请求而执行的工作的性质可能会有很大变化。使用使用者服务的单个实例可能导致该实例被请求淹没。或者，消息系统可能因来自应用程序的大量消息而过载。为了处理这种波动的工作负荷，系统可以运行使用者服务的多个实例。但是，这些使用者必须进行协调以确保每条消息仅传送给一个使用者。工作负荷还需要在使用者之间处于负载均衡状态，以防止实例成为瓶颈。

解决方案

使用消息队列来实现应用程序和使用者服务实例之间的信道。应用程序以消息的形式将请求发送到队列，使用者服务实例从队列接收消息并进行处理。此方法可让使用者服务实例的相同池处理来自应用程序实例的消息。该图说明了如何使用消息队列将工作分布到服务实例。此解决方案具有以下优点：

它提供了一个负载调节系统，可以处理应用程序实例发送的大幅度变化的请求量。队列充当应用程序实例和使用者服务实例之间的缓冲区。此缓冲区有助于最大程度地减轻对应用程序和服务实例的可用性和响应能力造成的影响。有关详细信息，请参阅基于队列的负载调控模式。处理需要长时间运行处理的消息时不会阻止使用者服务的其他实例同时处理其他消息。
它提高了可靠性。如果生成者直接与使用者通信，而不使用这种模式且不对使用者进行监视，则消息很可能丢失或未能处理（如果使用者失败）。在此模式中，消息不会发送到特定服务实例。失败的服务实例不会阻止生成者，并且任何工作服务实例都可处理消息。
它不需要使用者之间或生成者与使用者实例之间的复杂协调。消息队列可确保每条消息至少传送一次。
可缩放。如果应用了自动缩放，则消息数量波动时，系统可以动态地增加或减少使用者服务实例的数量。
如果消息队列提供事务读取操作，则可以提高复原能力。如果使用者服务实例读取和处理消息（作为事务操作的一部分），并且使用者服务实例失败，则该模式可以确保消息将返回到队列由另一使用者服务实例进行选取并处理。为了降低消息持续失败的风险，建议使用死信队列。

问题和注意事项

在决定如何实现此模式时，请考虑以下几点：

消息排序。不能保证使用者服务实例接收消息的顺序，且不一定反映创建消息的顺序。设计系统以确保消息处理是幂等的，因为这有助于消除对消息处理顺序的任何依赖。有关详细信息，请参阅 Jonathon Oliver 博客中的 Idempotency Patterns（幂等模式）。

Microsoft Azure 服务总线队列可通过消息会话对消息执行保证的先进先出顺序。有关详细信息，请参阅使用会话的消息传送模式。
为复原能力设计服务。如果系统专用于检测和重新启动失败的服务实例，则可能需要实现由服务实例作为幂等操作执行的处理，以尽量减少对多次检索和处理的单个消息的影响。
检测有害消息。格式不正确的消息或需要访问不可用资源的任务可能会导致服务实例失败。系统应阻止此类消息返回队列，并在其他位置捕获和存储这些消息的详细信息，以便在必要时对其进行分析。
处理结果。处理消息的服务实例与生成消息的应用程序逻辑完全分离，它们可能无法直接通信。如果服务实例生成必须传递回应用程序逻辑的结果，则此信息必须存储在两者都可访问的位置。为了防止应用程序逻辑检索不完整的数据，系统必须在处理完成时指示。

如果使用的是 Azure，工作进程可使用专用消息答复队列将结果传回应用程序逻辑。应用程序逻辑必须能够将这些结果与原始消息相关联。有关此方案的详细信息，请参阅 Asynchronous Messaging Primer（异步消息传送入门）。
缩放消息传送系统。在大规模解决方案中，单个消息队列可能不堪应付太多的请求，并且在系统中成为瓶颈。在这种情况下，请考虑对消息系统进行分区以将消息从特定生成者发送到特定队列，或者使用负载均衡在多个消息队列之间分发消息。
确保消息传送系统的可靠性。需要可靠的消息传递系统来保证在应用程序将消息放入队列之后它不会丢失。此系统对于确保所有消息至少传递一次至关重要。

何时使用此模式

在以下情况下使用此模式：

应用程序的工作负荷分为可以异步运行的任务。
任务是独立的且可并行运行。
工作量是多变的，因此需要可缩放的解决方案。
该解决方案必须提供高可用性，并且如果任务的处理失败，必须具有复原能力。

在以下情况下，此模式可能不起作用：

难以将应用程序工作负荷分成离散任务，或任务之间存在高度依赖性。
任务必须同步执行，且应用程序逻辑必须等待任务完成后才能继续。
必须以特定顺序执行任务。

某些消息传递系统支持会话，使生成者能够将消息组合在一起，并确保由相同的使用者进行处理。此机制可用于按优先级排列的消息（如果支持）以实现消息排序的形式，从生成者到单个使用者按顺序传送消息。

示例

Azure 提供服务总线队列和 Azure Functions 队列触发器，两者的结合是这种云设计模式的直接实现。 Azure Functions 通过触发器和绑定来与 Azure 服务总线集成。通过与服务总线集成，可以生成函数来使用发布者发送的队列消息。发布方应用程序会将消息发布到队列，而作为 Azure Functions 实现的使用者可以从该队列中检索消息并对其进行处理。为实现复原，服务总线队列允许使用者在从队列中检索消息时使用 PeekLock 模式；此模式实际上不会删除消息，而只是对其他使用者隐藏消息。 Azure Functions 运行时以 PeekLock 模式接收消息，如果函数成功完成，它会对消息调用 Complete；如果函数失败，它可能会调用 Abandon，这样，消息将再次可见，另一个使用者可以检索它。如果函数的运行时间长于 PeekLock 超时，则只要该函数正在运行，就会自动续订锁。 Azure Functions 可以根据队列深度进行横向扩展/缩减，所有函数实例充当队列的竞争使用者。如果创建了多个函数实例，它们都会通过独立拉取和处理消息来参与竞争。有关如何使用 Azure 服务总线队列的详细信息，请参阅 Service Bus Queues, Topics, and Subscriptions（服务总线队列、主题和订阅）。有关队列触发的 Azure Functions 的信息，请参阅 Azure Functions 的 Azure 服务总线触发器。以下代码演示了如何使用 ServiceBusClient 实例创建新消息并将其发送到服务总线队列。

private string serviceBusConnectionString = ...;
...

  public async Task SendMessagesAsync(CancellationToken  ct)
  {
   try
   {
    var msgNumber = 0;

    var serviceBusClient = new ServiceBusClient(serviceBusConnectionString);

    // create the sender
    ServiceBusSender sender = serviceBusClient.CreateSender("myqueue");

    while (!ct.IsCancellationRequested)
    {
     // Create a new message to send to the queue
     string messageBody = $"Message {msgNumber}";
     var message = new ServiceBusMessage(messageBody);

     // Write the body of the message to the console
     this._logger.LogInformation($"Sending message: {messageBody}");

     // Send the message to the queue
     await sender.SendMessageAsync(message);

     this._logger.LogInformation("Message successfully sent.");
     msgNumber++;
    }
   }
   catch (Exception exception)
   {
    this._logger.LogException(exception.Message);
   }
  }

以下代码示例演示了一个编写为 C# Azure 函数的使用者，它读取消息元数据并记录服务总线队列消息。请注意如何使用 ServiceBusTrigger 属性将它绑定到服务总线队列。

[FunctionName("ProcessQueueMessage")]
public static void Run(
    [ServiceBusTrigger("myqueue", Connection = "ServiceBusConnectionString")]
    string myQueueItem,
    Int32 deliveryCount,
    DateTime enqueuedTimeUtc,
    string messageId,
    ILogger log)
{
    log.LogInformation($"C# ServiceBus queue trigger function consumed message: {myQueueItem}");
    log.LogInformation($"EnqueuedTimeUtc={enqueuedTimeUtc}");
    log.LogInformation($"DeliveryCount={deliveryCount}");
    log.LogInformation($"MessageId={messageId}");
}

后续步骤

异步消息传送入门。消息队列是异步通信机制。如果使用者服务需要向应用程序发送回复，则可能需要实现某种形式的响应消息传送。异步消息传送入门提供了有关如何使用消息队列实现请求/回复消息传送的信息。
Autoscaling Guidance（自动缩放指南）。由于队列应用程序发布消息的长度不同，因此可能可以启动和停止使用者服务的实例。自动缩放有助于在峰值处理期间保持吞吐量。

云计算架构-设计模式

使用者竞争