消息传输、锁定和处置

消息代理（如服务总线）的最核心功能是将消息接受到队列或主题中以及保存它们以用于将来检索。 术语 发送 用于描述将消息传输到消息中转站的过程，而 接收 是指从中转站检索消息的过程。

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

同样，当中转站向客户端传输消息时，中转站和客户端都希望了解消息是已成功处理（可以删除消息），还是消息传递或处理失败（可能需要再次传递消息）。

使用任何支持的服务总线 API 客户端时，到服务总线中的发送操作总是需要显式确认，这意味着 API 操作将等待服务总线的确认结果到达，然后完成发送操作。

如果服务总线拒绝消息，则拒绝会包含错误指示器以及具有“tracking-id”的文本。 拒绝还包含有关是否可以在期望成功的情况下重试操作的信息。 在客户端中，此信息会被转化为异常并抛出到发送操作的调用方。 如果接受了消息，则操作以无提示方式完成。

高级消息队列协议 (AMQP) 是唯一支持 .NET Standard、Java、JavaScript、Python 和 Go 客户端的协议。 对于 .NET Framework 客户端，你可以使用服务总线消息传递协议 (SBMP) 或 AMQP。 使用 AMQP 协议时，消息传输和结算会进行管道处理，并且是异步方式。 建议使用异步编程模型 API 变体。

发送方可以快速连续地将多个消息置于线路上，而不必等待确认每个消息（使用 SBMP 协议或使用 HTTP 1.1 时会是这种情况）。 当相应消息在分区实体上被接受和存储时，或者在发送操作针对不同实体重叠时，这些异步发送操作将完成。 完成也可能不会按原始发送顺序进行。

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

如果应用程序会产生消息突发（此处使用普通循环说明），并且会等待每个发送操作完成，然后再发送下一个消息，则同步或异步 API 的情况相似，发送 10 个消息只会在 10 个连续的完整处置往返之后完成。

假设从本地站点到服务总线有 70 毫秒的 传输控制协议 (TCP) 往返延迟，并且服务总线接受和存储每个消息只需 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 客户端，服务总线还接受“预处置”传输。 预处理传输是一种即发即弃的操作，其结果无论如何都不会反馈给客户端，且消息在发送时即被视为已处理。 不向客户端进行反馈还意味着没有任何可操作数据可用于诊断，这表示此模式没有资格通过 Azure 支持获得帮助。

对于接收操作，服务总线 API 客户端启用两种不同的显式模式：接收并删除和扫视锁定。

接收并删除模式告知代理将它发送到接收客户端的所有消息都在发送时视为已处置。 这意味着在代理将消息置于线路上之后，它会立即被视为已使用。 如果消息传输失败，则消息会丢失。

此模式的优点是接收方无需对消息执行进一步操作，也不会由于等待处置结果而减慢速度。 如果各个消息中包含的数据具有较低值并且/或者只在很短时间内才有意义，则此模式是合理选择。

Peek-Lock模式告知消息代理，接收客户端希望明确地处理接收到的消息。 消息可供接收方进行处理，同时在服务中保持在排他锁下，以便其他竞争接收方无法看到它。 该锁的持续时间最初在队列或订阅级别进行定义，可以由拥有该锁的客户端通过 RenewMessageLockAsync 操作进行延长。 有关续订锁定的详细信息，请参阅本文中的续订锁部分。

锁定消息时，从相同队列或订阅进行接收的其他客户端可以接受锁并检索不处于活动锁下的下一个可用消息。 明确释放对消息的锁定或锁定过期时，消息可能会被置于检索顺序的最前面或靠近最前面的位置，以便重新传递。

当接收方多次释放消息或允许锁过期达到设定的次数 (最大传递计数) 时，系统会自动将消息从队列或订阅中删除，并将其放入关联的死信队列中。

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

如果接收客户端未能处理消息，但是希望重新传递消息，则可以通过为消息调用弃用 API 来显式要求立即释放并解锁消息，或者，也可以不执行任何操作，让锁结束。

如果接收客户端未能处理消息，并且知道重新传递消息和重试操作不起作用，则可以拒绝消息，这会通过为消息调用死信 API 而将其移动到死信队列中，因此还允许设置自定义属性（包括可以从死信队列中的消息检索的原因代码）。

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

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

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

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

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

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

• 一种特殊的结算情况是延期。 有关详细信息，请参阅消息延迟。
• 若要了解死信，请参阅死信队列。
• 若要大致了解有关服务总线消息的详细信息，请参阅服务总线队列、主题和订阅