如何主动处理Azure服务总线队列消息

Question

我们目前正在利用Azure服务总线来处理来自应用程序的各种消息。

我想知道实时处理这些消息的最佳方式是什么？

有没有一种方法可以在消息放入队列时自动执行脚本？

我只是在想，一定有比让一个单独的应用程序每分钟/30秒检查一次队列更好的方法。

谢谢各位

Answer 1

当涉及到基础架构代码时，我宁愿不写任何代码。毕竟，您最不希望看到的是基础架构代码中的一个bug，它会导致数据/消息丢失。

使用裸骨Azure服务总线的另一种选择是使用库为您提取所有代码。最终，您将声明您的消息—您的事件和命令—并且您拥有当出现消息时将被触发的处理程序。所有的消息泵送、队列创建、重试、错误处理、审计和事务（这些只是冰山一角）都有这样的框架。

至于使用哪种框架，有Nimbus和NServiceBus，可能还有其他。NServiceBus是一款商业产品，它附带了大量文档、devops、调试和可视化实用程序，以及您需要的额外付费支持。以下是如何使用Azure ServiceBus启动并运行NServiceBusendpoint：

var endpointConfiguration = new EndpointConfiguration("Endpoint1");
endpointConfiguration.SendFailedMessagesTo("error");

var transport = endpointConfiguration.UseTransport<AzureServiceBusTransport>();
var connectionString = Environment.GetEnvironmentVariable("AzureServiceBus.ConnectionString");

transport.ConnectionString(connectionString);
transport.UseTopology<ForwardingTopology>();

var endpointInstance = await Endpoint.Start(endpointConfiguration)
        .ConfigureAwait(false);

var message = new Message1
{
    Property = "Hello from Endpoint1"
};

await endpointInstance.Send(message).ConfigureAwait(false);

在接收方，您只需拥有一个处理程序类：

public class MyMessageHandler : IHandleMessages<Message1>
{
    public Task Handle(Message2 message, IMessageHandlerContext context)
    {
        //Do your task
        Console.WriteLine(message.Property);
        return Task.CompletedTask;
    }
}

另外，我为NServiceBus的特定软件制造商工作，但我使用了两种推荐的框架。你需要决定哪一个适合你。

Answer 2

您不需要根据计时器不断检查总线。

服务总线主题和订阅支持发布/订阅消息传递通信模型。

将消息发送到主题时，每个订阅都可以独立处理/处理该消息。

下面是一个C#示例，介绍如何从主题接收消息：

string connectionString =
    CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");

SubscriptionClient Client =
    SubscriptionClient.CreateFromConnectionString
            (connectionString, "TestTopic", "HighMessages");

// Configure the callback options.
OnMessageOptions options = new OnMessageOptions();
options.AutoComplete = false;
options.AutoRenewTimeout = TimeSpan.FromMinutes(1);

Client.OnMessage((message) =>
{
    try
    {
        // Process message from subscription.
        Console.WriteLine("\n**High Messages**");
        Console.WriteLine("Body: " + message.GetBody<string>());
        Console.WriteLine("MessageID: " + message.MessageId);
        Console.WriteLine("Message Number: " +
            message.Properties["MessageNumber"]);

        // Remove message from subscription.
        message.Complete();
    }
    catch (Exception)
    {
        // Indicates a problem, unlock message in subscription.
        message.Abandon();
    }
}, options);

以下是有关发布服务器订阅服务器型号的更多详细信息：

https://azure.microsoft.com/en-us/documentation/articles/service-bus-dotnet-how-to-use-topics-subscriptions/