标签: azureservicebus

我看过一些有关此主题的旧帖子，但没有任何最新内容且与较新的服务总线和 Azure Functions 堆栈相关，因此我想再次问这个问题。

在使用服务总线触发器的 Azure 函数上，有没有办法访问服务总线消息属性？据我所知，似乎只有消息正文字符串被传递到 Run 方法中。

同样，在使用服务总线输出绑定的 Azure 函数上，是否有办法将消息属性添加到出站服务总线消息？同样，据我所知，输出绑定似乎只接受消息正文字符串。

我们目前正在 Azure Function App 上运行负载测试，但吞吐量未达到我们的预期。

函数应用程序中有多种函数，但流量最多的函数是具有事件中心触发器的函数和具有使用来自启用会话的队列的消息的服务总线触发器的函数。

当系统处于负载状态时，会话启用队列中的消息将在队列中等待最多 10 分钟，直到由消费函数处理。

我知道 host.json 中有一些设置可以调整此行为，但它与我们的预期仍然相去甚远。

这是我们的host.json

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensions": {
    "serviceBus": {
      "prefetchCount": 100,
      "sessionHandlerOptions": {
        "autoComplete": true,
        "messageWaitTimeout": "00:00:30",
        "maxAutoRenewDuration": "00:55:00",
        "maxConcurrentSessions": 200
      },
      "batchOptions": {
        "maxMessageCount": 1000,
        "operationTimeout": "00:01:00",
        "autoComplete": true
      }
    }
  }
}

因此，我希望 Function App 能够同时处理最多 200 个会话，但事实上，尽管 Function Runtime 提供了大量实例，但大多数实例似乎都闲置在那里。所以对我来说，似乎还有另一个设置限制了功能应用程序的吞吐量。

我知道如果我们将函数拆分为单独的函数应用程序，将会提高性能，但由于两个函数的负载非常相似，我的计划是将此步骤推迟到稍后阶段，并且仍然通过单个函数应用程序获得可接受的吞吐量。

我们在 .NET Core 3.1 上使用 Azure Functions 3

• Microsoft.Azure.Functions.Extensions 1.1.0
• Microsoft.Azure.WebJobs.Extensions.ServiceBus 5.0.0 …

我在 Azure Function 中使用版本 5.2.0 的 Microsoft.Azure.WebJobs.Extensions.ServiceBus 包。根据此处的文档，新版本支持失败操作的自动重试。这是我的 host.json：

{
  "version": "2.0",
  "extensions": {
    "serviceBus": {
      "clientRetryOptions": {
        "mode": "Exponential",
        "tryTimeout": "00:02:00",
        "delay": "00:00:02.00",
        "maxDelay": "00:30:00",
        "maxRetries": 10
      }
    }
  }
}

除了时间值之外，它与文档中的完全相同。在我的由 a 触发的函数中，ServiceBusTrigger我在某些情况下抛出异常。我希望消息被延迟，如文档中所述。但它被放回到队列的头部并立即重试，没有任何延迟。我花了 2 个工作日来处理这个问题，混杂了各种版本、不正确的文档和我在网上找到的示例，我尝试了各种排列方式，但我就是无法让它发挥作用。

我再也见不到树木了。谁能看到我做错了什么吗？

我正在开发一个在 Dynamics 365 CRM 中执行多项操作的 Azure Function。我不完全理解 Azure Functions 并发的工作原理。我有一个消费计划，我的 Azure 函数内部有一个由服务总线消息触发的函数。

当我第一次测试它时，服务总线收到大约 200 条消息，并且应用程序开始同时处理大量消息，向 Dynamics 365 发出大量无法处理的请求。因此，在 Azure 门户中，我设法将最大实例数设置为 1，但该函数仍然一次处理许多消息。对此设置限制的最佳方法是什么？在 host.json 中使用 maxConcurrentCalls？在 host.json 中使用 maxConcurrentSessions？在应用程序配置中使用 WEBSITE_MAX_DYNAMIC_APPLICATION_SCALE_OUT 吗？

另外，将 maxConcurrentCalls 设置为 10 和 1 个函数实例或将其设置为 5 和 2 个函数实例有什么区别？

在我们在 Visual Studio 中本地开发的函数应用程序中，我们有两个运行环境：测试和开发。

现在，我们有一个local.settings.json和两个各自的特定环境，local.settings.<environment>.json 它们工作得很好 -但有一个例外：函数触发注释中使用的值。

它们似乎在实际过程开始之前就已经解决了local.settings.json。

它是 dotnet 6 v4，独立进程。该项目设置为读取local.settings.json 和 local.settings.<environment>.json。

相关文件如下所示：

lauchSettings.json

{
  "profiles": {
    "My.Project.Functions": {
      ...,
      "environmentVariables": {
        "AZURE_FUNCTIONS_ENVIRONMENT": "Test"
      }
    }
  }
}

local.settings.Test.json

{
    "myTopicName": "some-topic-for-test",
    "myServiceBusConnection": "Endpoint=sb://MyTestBusURI"
}

MyFunction.cs

public class MyFunction
{

    [Function(nameof(MyFunctionAsync))]
    public async Task MyFunctionAsync(
       [ServiceBusTrigger("%myTopicName%", Connection = "myServiceBusConnection")] // <-- error here
       MyMessageEntity myMessageEntity)
    {
        // resolving environment-specific values _here_ will work perfectly fine.
        await _someIrrelevantService.DoSomethingAsync(myMessageEntity);
    }
} …

我正在尝试连接到 Azure 服务总线，但遇到了问题。\n似乎无法连接，因为我总是收到此错误：

\n

Azure.Identity.AuthenticationFailedException：ClientSecretCredential 身份验证失败：未设置响应，请确保调用 SendAsync

\n

这是我正在尝试做的事情的代码片段

var adminClient = new ServiceBusAdministrationClient(\n    mySettings.AzureServiceBus.Endpoint,\n    new ClientSecretCredential(mySettings.AzureServiceBus.TenantId,\n        mySettings.AzureServiceBus.ClientId, mySettings.AzureServiceBus.ClientSecret));\n\nvar exists = await adminClient.SubscriptionExistsAsync(myTopic, mySubName).ConfigureAwait(false); \n

诊断来自AzureEventSourceListener：

[Informational] Azure-Identity: ClientSecretCredential.GetToken invoked. Scopes: [ https://servicebus.azure.net/.default ] ParentRequestId: 8cd8075d-fb77-4479-bf12-d612fff0de99\n[Informational] Azure-Identity: (False) MSAL 4.30.1.0 MSAL.Desktop Windows 10 Enterprise [02/01/2023 23:46:30 - ] MSAL MSAL.Desktop with assembly version \'4.30.1.0\'. CorrelationId(e1994990-3e32-43b9-83e5-55a325de668a)\n[Informational] Azure-Identity: (False) MSAL 4.30.1.0 MSAL.Desktop Windows 10 Enterprise [02/01/2023 23:46:30 - ] === AcquireTokenForClientParameters ===\nSendX5C: False\nForceRefresh: False\n\n[Informational] Azure-Identity: …

我是使用基于消息的体系结构的新手,例如Azure Service Bus主题/订阅功能.

我想知道如何最好地使用不同的消息类型.

例如,我说有两条消息.一个用于创建新客户,另一个用于删除客户.

我可以;

1. 创建两个主题.每个主题都有一个订阅.处理消息的代码将独立处理其消息类型.

2. 为客户创建一个主题.创建一个订阅以接收所有消息.处理消息的代码需要在处理之前确定消息类型.

3. 为客户创建一个主题.创建两个订阅,过滤消息类型.处理消息的代码将独立处理其消息类型.

我确信没有对错,但我会很感激在这方面有经验的人的一些意见.

非常感谢,

大卫

我正在尝试Azure Service Bus队列.我有以下代码:

队列发送:

string strConnectionString = ConfigurationManager.AppSettings["Microsoft.ServiceBus.ConnectionString"];
        var namespaceManager = NamespaceManager.CreateFromConnectionString(strConnectionString);
        if (!namespaceManager.QueueExists("Test"))
        {
            QueueDescription qD = new QueueDescription("Test");
            qD.DefaultMessageTimeToLive = new TimeSpan(05, 00, 00);
            qD.LockDuration = new TimeSpan(00, 02, 30);
            qD.MaxSizeInMegabytes = 5120;                
            namespaceManager.CreateQueue(qD);              
        }
        if (namespaceManager.QueueExists("Test"))
        {
            QueueClient client = QueueClient.CreateFromConnectionString(strConnectionString, "Test", ReceiveMode.PeekLock);
            var qMessage = Console.ReadLine();
            using (MemoryStream strm = new MemoryStream(Encoding.UTF8.GetBytes(qMessage)))
            {
                BrokeredMessage bMsg = new BrokeredMessage(strm);
                bMsg.MessageId = Guid.NewGuid().ToString();
                bMsg.TimeToLive = new TimeSpan(05, 00, 00);
                client.Send(bMsg);
                Console.WriteLine("Message sent");
            }
        }
        Console.ReadLine();

接收代码:

 string strConnectionString …

我要实现的是内部网络中的RabbitMQ客户端与Azure中运行的Azure Service Bus使用者之间的互操作性。

RabbitMQ客户端需要发布和订阅，Azure Service Bus使用者也需要发布和订阅-因此，我需要某种“双向代理”。我正在尝试实现的图表：

                                      +
                    Internal network  |  Azure
                                      |
                                      |
+--------+                            |                               +----------+
| Client +---+                        |                           +---+ Consumer |
+--------+   |                        |                           |   +----------+
             |                        |                           |
             | +-----------------+    |    +-------------------+  |
             +-+ RabbitMQ Broker +---------+ Azure Service Bus +--+
             | +-----------------+    |    +-------------------+  |
             |                        |                           |
+--------+   |                        |                           |   +----------+
| Client +---+                        |                           +---+ Consumer |
+--------+                            |                               +----------+
                                      |
                                      |
                                      |
                                      +

AFAIK，RabbitMQ代理和Azure Service Bus都可以执行AMQP 1.0。我已经看过了兔子铲插件，但是我认为这将只处理从内部客户端到Azure的消息发布，并且不允许客户端订阅Azure使用者发布的消息吗？还是我弄错了，铲子可以解决这个问题？

如果铲子不起作用，还有其他方法可以实现这一目标

我想从.NET Core 2.1应用程序内部使用Azure Service Bus。我熟悉Nuget软件包随附的SDK，Microsoft.Azure.ServiceBus并使用它目前正在写一个主题并从那里接收消息。

我现在想要的是为每个客户端在此主题上创建订阅。关键是我想使用创建过滤订阅SqlFilter。这意味着，我必须通过代码创建预订。

顺便说一句，据我所知，在代码中执行此操作的另一种方法是使用ARM模板进行部署，因为门户不允许我创建过滤的订阅。

我知道，这Microsoft.Azure.ServiceBus无法创建资源，因此我去了Nuget包Microsoft.Azure.Management.ServiceBus。因此，我现在可以通过如下代码创建新的订阅：

private static async Task CreateSubscriptionAsync()
{
    var context = new AuthenticationContext($"https://login.microsoftonline.com/{Configuration["AppSettings:AzureManagement:TenantId"]}");
    var token = await context.AcquireTokenAsync(
        "https://management.core.windows.net/",
        new ClientCredential(Configuration["AppSettings:AzureManagement:ClientId"], Configuration["AppSettings:AzureManagement:ClientSecret"]));
    var creds = new TokenCredentials(token.AccessToken);
    using (var sbClient = new ServiceBusManagementClient(creds)
    {
        SubscriptionId = VariableHelper.Configuration["AppSettings:AzureManagement:AzureSubscriptionId"]
    })
    {                
        var queueParams = new SBSubscription
        {
            LockDuration = TimeSpan.FromSeconds(10)
        };          
        await sbClient.Subscriptions.CreateOrUpdateAsync(
            Configuration["AppSettings:ServiceBus:ResourceGroup"],
            Configuration["AppSettings:ServiceBus:Namespace"],
            Configuration["AppSettings:ServiceBus:TopicName"],
            "mysub",
            queueParams);                
    }
}

这有效，因此我有一个新订阅，也可以删除。但是现在在哪里可以定义过滤器？该SBSubscription型不包含选项来定义过滤器，并且没有方法-过载sbClient.Subscriptions.CreateOrUpdateAsync。

我发现在代码中执行此操作的唯一方法是基于旧的Nuget WindowsAzure.ServiceBus（不适用于.NET …