Azure 服务总线限制

Question

我们有一个围绕服务总线构建的架构，它运行良好。我们目前还使用相同的名称空间服务总线通过定制的 Serilog 接收器（将被 datadog 替换）进行日志记录。

这导致我们在我们的测试环境中，偶尔会通过服务总线受到限制（我们是标准而不是高级）。我们知道这是我们这边的问题，正在努力解决。

但是，我想更新我们的代码以更优雅地处理限制异常，并可能在再次尝试之前添加延迟。

文档中不清楚的是异常详细信息：

https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-throttling

我们目前捕获了 MessagingException - 不清楚是否会抛出此消息。

有人用这种方式处理过节流异常吗？

目前的解决方法似乎是捕获错误，检查异常消息与异常消息（我认为查看错误代码是最不紧密耦合的实现）并将其放入专用代码来对其做出反应。

Answer 1

我们在服务总线中很少有异常，例如 QuotaExceededExceptions、ServerBusyExceptions、MessageSizeExceededExceptions、TransactionSizeExceededExceptions。

并且在某些节流条件下，会进行多次重试以确保最终传递消息。

使用 Azure Monitor，我们可以处理来自指标的限制： 以下是我经历的几个步骤：

1. 可以检查限制请求
2. 我们可以选择ServiceBusThrottling

您可以查看此blog 以了解如何使用服务总线的 Azure 函数处理节流调用。

下面是示例代码：

using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Host;
using Microsoft.ServiceBus.Messaging;
using System.Data.SqlClient;
using System.Text;
using System;

namespace BallardChalmers.BackgroundFunctions
{
    public static class BallardChalmersBackgroundTaskCreated
    {
        [FunctionName("BallardChalmersBackgroundTaskCreated")]
        public static void 
Run([ServiceBusTrigger("BallardChalmersBackgroundTaskCreated", AccessRights.Send, Connection = "ServiceBusConnection")]string queueMessage, TraceWriter log)
        {
            log.Info($"Service_TaskManager task added to queue with ID: {queueMessage}");
            try
            {
                string connectionString = System.Configuration.ConfigurationManager.AppSettings["SQLConnectionString"]; ;
                using (SqlConnection connection = new SqlConnection(connectionString))
                {
                    // Create wait period between 1 and 60 seconds
                    System.Random random = new System.Random();
                    int waitPeriod = random.Next(1, 60);

                    log.Info($"Wait period: {waitPeriod.ToString()}");
                    connection.Open();

                    StringBuilder sb = new StringBuilder();
                    sb.Append("INSERT FunctionsLog ");
                    sb.Append("SELECT '" + queueMessage + "','Started','" + DateTime.Now.ToString() + "','" + DateTime.Now.ToString() + "'," + waitPeriod.ToString());
                    String sql = sb.ToString();

                    log.Info($"Insert command: {sql}");
                    using (SqlCommand command = new SqlCommand(sql, connection))
                    {
                        command.ExecuteNonQuery();
                    }
                    log.Info($"Inserted to FunctionsLog");       

                    System.Threading.Thread.Sleep(1000 * waitPeriod);
                    log.Info($"Wait completed");

                    sb = new StringBuilder();
                    sb.Append("UPDATE FunctionsLog ");
                    sb.Append("SET [Status]='Completed', Updated='" + DateTime.Now.ToString() + "'");
                    sb.Append("WHERE ID='" + queueMessage + "'");
                    sql = sb.ToString();
                    log.Info($"Update command: {sql}");
                    using (SqlCommand command = new SqlCommand(sql, connection))
                    {
                        command.ExecuteNonQuery();
                    }
                    log.Info($"Updated to FunctionsLog");
                }
            }
            catch (SqlException e)
            {
                log.Error($"Error writing to database: {e.Message + ":::" + e.StackTrace}");
            }

        }
    }
}