标签: azureservicebus

我有一个在 Docker 容器中运行的 Azure 函数。该功能需要使用Azure服务总线。所以我的函数是使用属性“ServiceBusAccount”来获取服务总线连接字符串

    [FunctionName("MyFunc")]
    [ServiceBusAccount("ServiceBusConnectionString")]
    public async Task MyFunc(...)
    ...

在配置生成器中，我运行以下代码以从 Azure Key Vault 获取一些值（包括服务总线连接字符串）：

     var configBuilder = new ConfigurationBuilder();

     var config = configBuilder.AddAzureKeyVault($"https://{config["keyVaultName"]}.vault.azure.net", config["keyVaultClientId"], config["keyVaultSecret"])
                .Build();

问题是，当主机启动时，我首先收到警告：

“警告：找不到名为“ServiceBusConnectionString”的值”

然后在我的函数被发现后我得到了这个错误：

发生主机错误。Microsoft.Azure.WebJobs.ServiceBus：Microsoft Azure WebJobs SDK ServiceBus 连接字符串“ServiceBusConnectionString”丢失或为空

我的理解是，主机环境正在尝试在启动函数之前以及在项目的启动类中调用配置生成器之前获取服务总线连接字符串。

有没有办法在运行时设置服务总线连接字符串？

编辑

如果您使用的是 Azure DevOps，这应该有助于解决此问题https://daniel-krzyczkowski.github.io/How-to-inject-Azure-Key-Vault-secrets-in-the-Azure-DevOps-CICD-pipelines /

我有一个 Azure 服务总线触发了 Azure 函数。当我在本地运行 Azure Function 时，它会启动 16 个线程并在每个线程中获取 16 条消息。我如何配置它，使其只运行一条消息，以便我可以调试它，而不会在同一个断点上命中 16 次？

我尝试在 host.json 文件中设置配置（如下所示），以便一次仅从 Azure 服务总线拾取 1 条消息，但这没有帮助。

{
  "version": "2.0",
  "extensions": {
    "serviceBus": {
      "prefetchCount": 100,
      "messageHandlerOptions": {
        "autoComplete": false,
        "maxConcurrentCalls": 1,
        "maxAutoRenewDuration": "00:55:00"
      }
    }
  }
}

编辑 1：我目前所做的是通过包含正文中消息输入的 http 请求触发函数的管理端点。这样做的问题是，http 请求正文必须包含 {"input":"{}"} 并且我每次都必须花时间使用转义双引号创建有效的 json。如果我能够将函数配置为从服务总线主题一次运行单个消息，就会容易得多。

我的队列接收器定期收到Microsoft.Azure.ServiceBus.ServiceBusException（下面的消息已删除敏感信息）。SAS 键具有发送/侦听访问权限，并且随着处理正常进行，该错误似乎无关紧要。然而，该消息在我的仪表板中造成了信噪比问题（每天收到 10-70 个错误）。关于为什么会发生这种情况有什么想法吗？侦听器正在 Azure 应用服务中运行，但我认为这并不重要。我已经调整了重试逻辑，以使用 RetryExponential 进行 1 秒到 1 分钟的退避，重试 5 次。

请求SDK开发者指导

套餐

Net Core 3.1

Microsoft.Azure.ServiceBus, Version=4.1.3.0, Culture=neutral, PublicKeyToken=7e34167dcc6d6d8c

错误信息

链接“xxx;xxx:xxx:xxx:source(address:xxx):xxx”被强制分离。代码：RenewToken。详细信息：未经授权的访问。执行此操作需要“监听”声明。资源：'sb://xxx.servicebus.windows.net/xxx'.. TrackingId:xxx，SystemTracker:xxx，时间戳：2020-04-27T09:36:04 链接 'xxx;xxx:xxx:xxx:source (地址:xxx):xxx' 被强制分离。代码：RenewToken。详细信息：未经授权的访问。执行此操作需要“监听”声明。资源：'sb://xxx.servicebus.windows.net/xxx'.. TrackingId：xxx，SystemTracker：xxx，时间戳：2020-04-27T09：36：04

来源

internal delegate TClient ClientFactory<out TClient>(string connectionString, string entityPath,
    RetryPolicy retryPolicy);

internal delegate Task OnMessageCallback<in TMessage>(TMessage message,
    CancellationToken cancellationToken = default) where TMessage : ICorrelative;

internal sealed class ReceiverClientWrapper<TMessage> : IReceiverClientWrapper<TMessage>
    where TMessage : ICorrelative
{
    // ReSharper disable once StaticMemberInGenericType
    private static readonly Regex TransientConnectionErrorRegex = …

我正在尝试为 Azure 服务总线中的消息计数创建图表，我看到有两个选项

i . Count of Messages in Queue/Topic
ii. Count of Active messages in Queue/Topic

这两者有什么区别？

我正在使用 Newtonsoft.Json 反序列化对象。我收到以下异常：

Newtonsoft.Json.JsonSerializationException
  HResult=0x80131500
  Message=Error converting value {null} to type 'System.DateTime'. Path 'StartDate', line 1, position 62.
  Source=Newtonsoft.Json
  
Inner Exception 1:
InvalidCastException: Null object cannot be converted to a value type.

原始 json 具有空值：

{
  "StartDate": null,
  "EndDate": null
}

但我向 JsonConvert.DeserializeObject 提供设置以避免空值，如此处和此处所述：

var convertedMessage = JsonConvert.DeserializeObject<T>(
                Encoding.UTF8.GetString(message.Body),
                new JsonSerializerSettings { NullValueHandling = NullValueHandling.Ignore });

知道为什么它一直抛出这个异常吗？

如果提供了实际的日期值，则代码运行良好。

有关代码的更多背景信息，message.Body 是通过服务总线（类 Microsoft.Azure.ServiceBus.Message）接收的消息的正文。对其应用 GetString 方法会返回与发送的消息中相同的字符串。

可运行的代码示例：

using System;
using System.Text;
using Microsoft.Azure.ServiceBus;
using Newtonsoft.Json;


namespace ConsoleApp1
{
    public …

1. 事件网格主题和事件网格系统主题有什么区别。

当我选择资源类型“事件网格自定义主题”时，将创建事件网格主题类型。

但是当我使用 Service Bus Premium->Event Registration 时。正在创建事件网格系统主题类型。

问题是哪一种更有效以及使用哪一种。因为我使用端点类型 Webhook并在消息发布到我的服务总线队列/主题时调用我的逻辑应用程序事件。

而在EventGrid 自定义主题中，我必须在逻辑应用程序中使用事件网格触发器，并且会自动创建事件自定义主题。

请告诉我这两者之间的区别以及何时使用哪个以及哪个更有效

我的 host.json 中有这些，但每次我运行该函数时，它都会并行运行比 1 更多的线程（队列中有消息）

{
  "version": "2.0",
  "extensions": {
    "serviceBus": {
      "prefetchCount": 1,
      "messageHandlerOptions": {
        "maxConcurrentCalls": 1
      }
    }
  }
}

我的职能

 [FunctionName(nameof(ResourceEventProcessorFunction))]
    public async Task Run([ServiceBusTrigger("%TopicName%", "%SubscriptionName%", Connection = "ServiceBusConnection", IsSessionsEnabled = true)]Message message, IMessageSession messageSession, ILogger log)

我正在将 Azure Functions v3 从 迁移Newtonsoft.Json到System.Text.Json并尝试让camelCase 在全球范围内工作。

对于这些：

• 信号R
• 服务总线输出绑定
• 宇宙数据库

我能够在JsonSerializerOptions全局范围内显式传递或设置它（SignalR），但我无法为IAsyncCollector.

这是我的代码：

[FunctionName(nameof(SampleFunction))]
public async Task Run(
    [ServiceBusTrigger(ServiceBusQueue.QueueA)] string json,
    [ServiceBus(ServiceBusQueue.QueueB)] IAsyncCollector<Delivery> queueB,
    [ServiceBus(ServiceBusQueue.QueueC)] IAsyncCollector<Driver> queueC,
    ExecutionContext context)
{
        // ... do some work

        await queueB.AddAsync(objectB).ConfigureAwait(false);

        // ... do some more work

        await queueC.AddAsync(objectC).ConfigureAwait(false);
}

objectB最终objectC乘坐服务巴士而不是骆驼箱。作为解决方法，我在接收函数上将属性名称设置为不区分大小写。

PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
PropertyNameCaseInsensitive = true

知道如何使用IAsyncCollector驼峰案例进行序列化吗？

如果不指定 ProcessMessageAsync 和 ProcessErrorAsync，则无法使用 ServiceBusProcessor。第一个方法是什么，很清楚，但我不确定在 ProcessErrorAsync 中做什么？以下方法相同吗？

var client = new ServiceBusClient(connectionString);
var processor = _client.CreateProcessor(queueName);

processor.ProcessMessageAsync += async arg =>
{
    try
    {
        //process message
    }
    catch (Exception ex)
    {
        Console.WriteLine(ex.Message);
    }
};

processor.ProcessErrorAsync += arg =>
{
    return Task.CompletedTask;
};

await _processor.StartProcessingAsync(cancellationToken);

和

var client = new ServiceBusClient(connectionString);
var processor = _client.CreateProcessor(queueName);

processor.ProcessMessageAsync += async arg =>
{
    //process message
};

processor.ProcessErrorAsync += arg =>
{
    Console.WriteLine(ex.Message);
    return Task.CompletedTask;
};

await _processor.StartProcessingAsync(cancellationToken);

当我使用服务总线生产者代码（如下）在 Azure 门户和我的应用程序中创建计划服务总线消息时，我会收到一个序列号。我将其保存在我的数据库中。

问题- 当我的服务总线使用者代码由计划消息出列触发时，序列号与服务总线生产者代码和通过 Azure 门户最初提供给我的序列号不同。

如图所示，其中“13”是 Azure 门户屏幕中显示的序列号。

这是接收预定消息的代码，您可以看到序列号不同！

这是我的消费者代码（不认为这很重要）

    private async Task MessageHandler(ProcessMessageEventArgs args)
    {
        string body = args.Message.Body.ToString();
        JObject jsonObject = JObject.Parse(body);
        var eventStatus = (string)jsonObject["EventStatus"];

        await args.CompleteMessageAsync(args.Message);

        // fetch row here by sequence number
        // edit some data from entity, then save
        int result = await dbContext.SaveChangesAsync();

    }

这是我的生产者代码

    public async Task<long> SendMessage(string messageBody, DateTimeOffset scheduledEnqueueTimeUtc)
    {
        await using (ServiceBusClient client = new ServiceBusClient(_config["ServiceBus:Connection"]))
        {
            ServiceBusSender sender = client.CreateSender(_config["ServiceBus:Queue"]);

            ServiceBusMessage message = new ServiceBusMessage(messageBody); …