消息传输、锁定和处置

消息中转站（如Service Bus）的核心功能是将消息接受到队列或主题中，并保存消息以供以后检索。 术语 发送 用于描述将消息传输到消息中转站的过程，而 接收 是指从中转站检索消息的过程。

当客户端发送消息时，它通常想知道中转站是否已正确传输并接受消息，或者是否发生了某种错误。 这种肯定或否定确认可以让客户端和中转站理解消息传输状态。 因此，它被称为处置。

同样，当中转站将消息传输到客户端时，中转站和客户端希望确定消息是否已成功处理并可以删除，或者消息传递或处理是否失败，因此可能需要再次传递消息。

处置发送操作

使用任何受支持的 Service Bus API 客户端时，发送到 Service Bus 的操作始终显式完成。 这意味着，此API操作在完成发送操作之前会等待Service Bus的接收结果到达。

如果Service Bus拒绝消息，将包含一个错误指示器和含有跟踪 id的错误信息。 拒绝还包含有关是否可以在期望成功的情况下重试操作的信息。 客户端将此信息转换为异常，并将其抛出给发送操作的调用方。 如果接受了消息，则操作以无提示方式完成。

高级消息队列协议（AMQP）是 .NET标准、Java、JavaScript、Python 和 Go 客户端唯一支持的协议。 对于 .NET Framework 客户端，可以使用Service Bus消息传送协议（SBMP）或 AMQP。 使用 AMQP 协议时，消息传输和结算会进行管道处理，并且是异步方式。 建议使用异步编程模型 API 变体。

发送方可以在网络上快速连续地传输多个消息，而无需等待确认每个消息，尽管在 SBMP 协议或 HTTP 1.1 中会这样。 当相应消息在分区实体上被接受和存储时，或者在发送操作针对不同实体重叠时，这些异步发送操作将完成。 完成也可能不会按原始发送顺序进行。

用于处理发送操作结果的策略可能会对应用程序形成直接的显著性能影响。 本部分中的示例采用 C# 编写，适用于Java期货、Java mono、JavaScript 承诺和其他语言的等效概念。

如果应用程序生成突发消息（此处用简单的循环展示）并在每次发送操作完成后再发送下一条消息，无论是同步还是异步API格式，发送10条消息仅在10个连续的完整往返行程完成后才能完成确认。

假设从本地站点到Service Bus的传输控制协议（TCP）往返延迟为 70 毫秒，Service Bus只需 10 毫秒即可接受和存储每个消息，则以下循环至少需要 8 秒，而不计算有效负载传输时间或潜在的路由拥塞影响：

for (int i = 0; i < 10; i++)
{
    // creating the message omitted for brevity
    await sender.SendMessageAsync(message);
}

如果应用程序即时连续地启动 10 个异步发送操作，并等待其分别完成，则会叠加这 10 个发送操作的往返时间。 10 个消息会立即连续传输，甚至可能共享 TCP 帧，并且总体传输持续时间在很大程度上取决于将消息传输到代理所需的网络相关时间。

进行与前面循环相同的假设，以下循环的总重叠执行时间可能刚好低于一秒：

var tasks = new List<Task>();
for (int i = 0; i < 10; i++)
{
    tasks.Add(sender.SendMessageAsync(message));
}
await Task.WhenAll(tasks);

请务必注意，所有异步编程模型都使用某种形式的基于内存的隐藏工作队列来保存挂起的操作。 发送 API 返回时，发送任务会在该工作队列中排队，但只有当轮到任务执行时，协议步骤才会开始。 对于倾向于推送突发消息且可靠性存疑的代码，要注意不要一次性发送过多的“在途”消息，因为所有已发送的消息在传送到网络之前都会占用内存。

如以下 C# 代码片段中所示，信号灯是可在需要时启用这类应用程序级别限制的同步对象。 信号灯的这种用法允许同时最多发送 10 个消息。 10 个可用信号灯锁中的一个会在发送之前采用，然后在发送完成时释放。 循环的第 11 次执行会等待以前发送操作中的至少一个完成，然后使其锁可用：

var semaphore = new SemaphoreSlim(10);

var tasks = new List<Task>();
for (int i = 0; i < 10; i++)
{
    await semaphore.WaitAsync();

    tasks.Add(sender.SendMessageAsync(message).ContinueWith((t)=>semaphore.Release()));
}
await Task.WhenAll(tasks);

应用程序绝不应以“不关心结果”的方式启动异步发送操作，而不检索该操作的结果。 这样做可以加载内部和不可见任务队列，直到内存耗尽，并阻止应用程序检测发送错误：

for (int i = 0; i < 10; i++)
{
    sender.SendMessageAsync(message); // DON'T DO THIS
}

使用低级别 AMQP 客户端时，Service Bus 还接受“预先确认”传输。 预处理传输是一种即发即弃的操作，其结果无论如何都不会反馈给客户端，且消息在发送时即被视为已处理。 客户端缺乏反馈也意味着没有可用于诊断的可作数据，这意味着此模式不符合通过Azure support获得帮助的条件。

处置接收操作

对于接收作，Service Bus API 客户端支持两种显式模式：Receive-and-Delete 和 Peek-Lock。

ReceiveAndDelete

接收和删除模式指示代理将它发送到接收客户端的所有消息在发送时视为已确认。 一旦代理将消息放到线路上，消息就会被消费。 如果消息传输失败，则消息会丢失。

接收方不需要对消息采取任何进一步的动作，也不会因等待结算结果而被延缓。 如果单个消息中包含的数据具有较低的值或仅在非常短的时间内有意义，则此模式是一种合理的选择。

PeekLock

Peek-Lock模式告知消息代理，接收客户端希望明确地处理接收到的消息。 消息可供接收方处理，而服务持有独占锁，以便其他竞争接收器看不到它。 队列或订阅最初定义锁的持续时间。 拥有锁的客户端可以使用 RenewMessageLockAsync 来扩展它。 有关续订锁的详细信息，请参阅本文中的 “续订锁 ”部分。

锁定消息后，从同一队列或订阅接收的其他客户端可以接受锁定，并检索未处于活动锁定状态的下一个可用消息。 当显式释放消息上的锁或锁过期时，该消息会返回到检索顺序中的最前面进行重新投递。

当接收方重复释放消息或让锁定时间超过定义的次数（最大传递计数），消息将自动从队列或订阅中移除，并放入关联的死信队列中。

当接收客户端调用消息的完成 API 时，它会使用肯定确认启动对已收到消息的处置。 它向中转站指示消息已成功处理，并从队列或订阅中删除该消息。 代理会使用指示是否可以执行处置的回复来回复接收者的处置意向。

当接收客户端无法处理消息但希望重新传送消息时，它可以通过调用 消息的“放弃 API”来显式请求释放和解锁消息。 或者，它什么都不做，让锁失效。

如果接收客户端无法处理消息并知道重新传送消息并重试作将无济于事，则它可能会拒绝该消息。 通过在消息上调用 DeadLetter API，该消息将进入死信队列。 此 API 还允许设置自定义属性，并且可以包括一个原因代码，该代码可以随同死信队列中的消息一起检索。

一种特殊的处置情况是延迟。 有关详细信息，请参阅 消息延迟 文章。

Complete、DeadLetter 或 RenewLock 操作可能因网络问题而失败，例如，锁定状态过期，或者存在其他阻止处置的服务端条件。 在后面一种情况下，服务会发送否定确认，该确认会在 API 客户端中表现为异常。 如果原因是网络连接中断，则会删除该锁，因为Service Bus不支持恢复其他连接上的现有 AMQP 链接。

如果 Complete 失败（通常发生在消息处理工作结束时，某些情况下是在处理工作进行几分钟后），接收应用程序可以决定是否保留当前工作状态，在第二次传递同一消息时忽略它，或者在重新传递消息时选择舍弃工作结果并重试。

标识重复消息传递的典型机制是通过检查 message-id。 发送方可以并应将message-id设置为唯一值，可能与发起进程的标识符一致。 作业调度程序可能会将 message-id 设置为它尝试分配给工作者的作业标识符。 如果作业已完成，工作人员将忽略作业分配的第二次出现。

设计幂等消息处理变得至关重要。 有关在分布式系统中实现幂等性的实用技术和示例，请参阅本指南：幂等是什么意思？

续订锁

锁定持续时间的默认值为“1 分钟”。 可以在 队列 或 订阅 级别为锁定持续时间指定不同的值。 拥有该锁的客户端可以使用接收方对象上的方法来续订消息锁。 您还可以使用自动锁续功能，指定要保持锁续的时间。

将锁定持续时间设置为高于正常处理时间的值，因此无需续订锁。 最大值为 5 分钟，因此，如果需要更长的时间，则需要延长锁定。 锁定持续时间超过所需时间也会产生一些影响。 例如，当客户端停止工作时，消息仅在锁定持续时间过后才会再次可用。

后续步骤

• 一种特殊的处置情况是延迟。 有关详细信息，请参阅消息延迟。
• 若要了解死信，请参阅死信队列。
• 若要详细了解一般Service Bus消息传送，请参阅 Service Bus 队列、主题和订阅。