从Microsoft.Azure.ServiceBus迁移到Azure.Messaging.ServiceBus

Question

我正在尝试更新这个函数，使用Azure.Messaging.ServiceBus代替Microsoft.Azure.ServiceBus，但是似乎找不到任何相关资源。有人知道如何使用这个包发送主题消息吗？

旧的函数是：

  public async Task SendMessageToServiceBusTopic<T>(T request, string topicSubName, string submissionNumber)
    {
        ServiceBusConnectionStringBuilder serviceBusConnectionStringBuilder =
            new ServiceBusConnectionStringBuilder(settings.ServiceBusConnectionString)
            {
                EntityPath = settings.ServiceBusTopic
            };

     
            TopicClient topicClient = new TopicClient(serviceBusConnectionStringBuilder);

            byte[] bytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(request));

            await topicClient.SendAsync(new Message(bytes)
            {
                CorrelationId = context.CorrelationId,
                Label=topicSubName,
                UserProperties = { new KeyValuePair<string, object>("TrackingId", submissionNumber) }
            });          
    }

到目前为止，我已经成功地完成了：

我是否朝着正确的方向前进？

  public async Task SendMessageToServiceBusTopic<T>(T request, string topicSubName, string submissionNumber)
    {
        ServiceBusClient client = new ServiceBusClient(settings.ServiceBusConnectionString);
        ServiceBusSender s = client.CreateSender(settings.ServiceBusTopic);


            byte[] bytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(request));
            await s.SendMessageAsync(new ServiceBusMessage(bytes));
      }

- azureLover

Answer 1

虽然您可以每次构造Service Bus客户端，但并不理想。假设您正在使用最新的In-Proc SDK，则可以使用以下选项之一：

[FunctionName("PublishToTopic")]
public static async Task Run(
    [TimerTrigger("0 */5 * * * *")] TimerInfo myTimer,
    [ServiceBus("<topic_name>", Connection = "<connection_name>")] IAsyncCollector<ServiceBusMessage> collector)
{
    await collector.AddAsync(new ServiceBusMessage(new BinaryData($"Message 1 added at: {DateTime.Now}")));
    await collector.AddAsync(new ServiceBusMessage(new BinaryData($"Message 2 added at: {DateTime.Now}")));
}

或者，

[FunctionName("PublishWithSender"]
public static async Task Run(
    [TimerTrigger("0 */5 * * * *")] TimerInfo myTimer,
    [ServiceBus("<topic_name>", Connection = "<connection_name>")] ServiceBusSender sender)
{
    await sender.SendMessagesAsync(new[]
    {
        new ServiceBusMessage(new BinaryData($"Message 1 added at: {DateTime.Now}")),
        new ServiceBusMessage(new BinaryData($"Message 2 added at: {DateTime.Now}"))
    });
}

对于Isolated Worker SDK，情况有些不同。请参阅此篇文章获取详细信息。

- Sean Feldman

Answer 2

我曾经和你一样处于同样的情况（尝试迁移到 Azure.Messaging.ServiceBus，这是微软推荐的 NuGet）。

你可以查看他们在 Github 上的迁移指南，以了解如何使过渡更加顺畅： https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/servicebus/Azure.Messaging.ServiceBus/MigrationGuide.md

/!\ IQueueClient 和 ITopicClient（两者都来自 Microsoft.Azure.ServiceBus）已合并为一个名为 ServiceBusSender 的单个对象。

为什么？为了让我们的生活更轻松！实际上，在幕后，ServiceBusSender 现在负责发送消息的过程，特别是因为如果我没记错的话，队列和主题不能具有相同的名称。

以下是我使用新库发送消息的示例代码：

    /// <summary>
    /// Send a message by serializing as JSON the object in input. 
    /// </summary>
    /// <param name="queueOrTopicName">Name of the queue or topic where the message will be sent.</param>
    /// <param name="messageToSend">Any C# object</param>
    /// <returns></returns>
    public async Task SendMessage(string queueOrTopicName, object messageToSend)
    {
        //ServiceBusSender should not be disposed (according to the documentation, Github, etc.)
        await using ServiceBusClient client = new ServiceBusClient(connectionString: "Your-ServiceBus-ConnectionString");
        await using ServiceBusSender sender = client.CreateSender(_queueOrTopicName);

        // create a message that we can send. UTF-8 encoding is used when providing a string.
        ServiceBusMessage message = BuildServiceBusMessage(messageToSend);

        // Finally send the message
        await sender.SendMessageAsync(message);
    }

    private ServiceBusMessage BuildServiceBusMessage<T>(T entity)
    {
        string serializedMessage = JsonConvert.SerializeObject(entity); // Still using Newtonsoft.Json but I don't see any obstacles of using System.Text.Json.
        ServiceBusMessage message = new ServiceBusMessage(serializedMessage)
        {
            MessageId = Guid.NewGuid().ToString(),
            ContentType = "application/json; charset=utf-8",
        };

        return message;
    }

如果您有任何需要进行依赖注入（例如重用相同的ServiceBusClient对象并避免为要发送的每条消息实例化ServiceBusClient），您可以参考我本周发现的这个stackoverflow链接： 如何注册ServiceBusClient以进行依赖注入？

- moueidat

---

ServiceBusClient及其实例化的“AMQP链接”对象，例如从CreateSender创建的ServiceBusSender，必须进行处理以避免句柄泄漏。文档对此解释不够清晰，而是侧重于重复连接和链接创建的性能影响。这里的示例在必要时处理了客户端和发送器，但仍会由于不断重新连接而导致过多的网络流量。无法分配更多句柄。句柄的最大数量为4999。（QuotaExceeded）。 - mkjeldsen

@mkjeldsen ServiceBusClient实际上没有实现IDisposable，因此我认为使用它作为静态对象以避免重新实例化是必要的。 - Dave Friedel

@DaveFriedel，“ServiceBusClient”和“ServiceBusSender”实现了“IAsyncDisposable”（但不是“IDisposable”）。由于这个原因，这个示例出现了编译错误，但我已经进行了更正。无论如何，关键是要管理句柄的生命周期以避免泄漏——全局单一的“静态”实例是一种方法，但这种机制强制从用户中删除了很多控制，并且在机械上也不比依赖显式生命周期管理的替代方案优越。 - mkjeldsen

澄清一下：问题中的第二个示例是不正确的。_本答案中的示例是正确的但效率低下_；一个高效的实现应该包括一些机制来重用客户端而不会最终丢弃它们。Sean Feldman的答案是针对另一个问题的。 - mkjeldsen

Answer 3

你正在朝着正确的方向前进。由于使用的是不同的库来处理相同的服务（Azure Service Bus），因此没有简单的迁移工具或示例。

- Thiago Custodio

---

还有其他需要吗，还是可以了？ - azureLover