Azure service bus Topic基本用法

我們在升級一個POS系統的時候，決定使用微軟公有雲計算平台下的Azure ServiceBus 進行POS客戶端與服務器的交互。

本文主要時作者在學習使用 Azure SDK for .NET 操作由世紀互聯運營的 中國區Azure 上的 Service Bus。

---

 

目錄
一、安裝AzureServiceBus程序集

二、在Portal創建命名空間

三、通過代碼創建Topic

四、通過代碼創建訂閱

五、創建並發送消息

六、消費消息

---

  

1.通過nuget安裝程序集

using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;

using Microsoft.Azure;

2. 在Portal創建 service bus 命名空間：

 

只有標准級別可以使用主題（Topic），因此創建命名空間時，請選擇標准；

也可以創建后，在縮放選項卡里調整為標准。

 

3.創建主題，可以通過portal創建，也可以通過代碼創建：

 

 通過代碼創建主題：

// Create the topic if it does not exist already.
            string connectionString =
                CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");

            var namespaceManager =
                NamespaceManager.CreateFromConnectionString(connectionString);

            if (!namespaceManager.TopicExists("TestTopic"))
            {

                //默認的createtopic方法
                 namespaceManager.CreateTopic("TestTopic");


                // 通過TopicDescription構建一個重載
                //TopicDescription td = new TopicDescription("TestTopicCustomer");
                //td.MaxSizeInMegabytes = 5120;
                //td.DefaultMessageTimeToLive = new TimeSpan(0, 1, 0);
                //if (!namespaceManager.TopicExists("TestTopicCustomer"))
                //{
                //    namespaceManager.CreateTopic(td);
                //}

            }

 

4.創建訂閱：

下面的代碼演示創建了三個訂閱，其中Product和Customer訂閱增加了一個SqlFilter，即如果發送消息時，

 message.Properties["Entity"] = "Product"，則消息會由Product訂閱處理；

 message.Properties["Entity"] = "Customer"，則消息會由Customer訂閱處理；

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

            var namespaceManager =
                NamespaceManager.CreateFromConnectionString(connectionString);

            if (!namespaceManager.SubscriptionExists("TestTopic", "AllMessages"))
            {
                namespaceManager.CreateSubscription("TestTopic", "AllMessages");
            }

            Console.WriteLine("AllMessages done");


            // Create a "ProductMessagesFilter" filtered subscription.
            SqlFilter ProductMessagesFilter =
               new SqlFilter("Entity='Product'");

            namespaceManager.CreateSubscription("TestTopic",
               "Product",
               ProductMessagesFilter);

            Console.WriteLine("Product done");

            // Create a "CustomerMessagesFilter" filtered subscription.
            SqlFilter CustomerMessagesFilter =
               new SqlFilter("Entity='Customer'");

            namespaceManager.CreateSubscription("TestTopic",
               "Customer",
               CustomerMessagesFilter);

            Console.WriteLine("Customers done");

 

5.創建並發送普通消息：

下面創建了5個product和3個customer並發送了消息，

 BrokeredMessage message = new BrokeredMessage(product);

 BrokeredMessage message = new BrokeredMessage(customer);

Product有如下設置：message.Properties["Entity"] = "Product";

Customer有如下設置：message.Properties["Entity"] = "Customer";

則 product 訂閱會有5條消息，customer訂閱會有3條消息，另外，Allmessages沒有任何sqlfilter，則該訂閱會有5+3=8條消息。

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

            TopicClient Client =
                TopicClient.CreateFromConnectionString(connectionString, "TestTopic");

           // Client.Send(new BrokeredMessage());


            //Add 5 Products
            var product = new Model.Product();

            for (int i = 0; i < 5; i++)
            {

                product.ProductID ="ProductID"+ i.ToString();

                // Create message, passing a Product  for the body.
                BrokeredMessage message = new BrokeredMessage(product);

                // Set additional custom app-specific property.
                message.Properties["Entity"] = "Product";

                //ScheduledMessage
                //message.ScheduledEnqueueTimeUtc = DateTime.Now.AddMinutes(1);

                // Send message to the topic.
                Client.Send(message);
            }


            //Add 3 Customers
            var customer = new Model.Customer();

            for (int i = 0; i < 3; i++)
            {

                customer.CustomerID = "CustomerID"+i.ToString();

                // Create message, passing a Customer message for the body.
                BrokeredMessage message = new BrokeredMessage(customer);

                // Set additional custom app-specific property.
                message.Properties["Entity"] = "Customer";

                // Send message to the topic.
                Client.Send(message);
            }

發送消息后的效果：

 創建定時消息：

設置消息的如下屬性，則消息在發送成功后延時1分鍾后才會出現在上圖中，也才可以被訂閱消費。

message.ScheduledEnqueueTimeUtc = DateTime.Now.AddMinutes(1);

6.消費消息：

不同的訂閱可以取不同訂閱里的消息進行處理，處理完成后，標記message.Complete()即可從訂閱里清除消息。

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

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

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

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

            // 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("Messages");
               
                    Console.WriteLine("MessageID: " + message.MessageId);
                    Console.WriteLine("Message EntityType=: " +
                        message.Properties["Entity"]);


                    if(message.Properties["Entity"].ToString()=="Customer")
                    {
                        Console.WriteLine("CustomerID=" + message.GetBody<Customer>().CustomerID.ToString());
                    }
                    if (message.Properties["Entity"].ToString() == "Product")
                    {
                        Console.WriteLine("ProductID=" + message.GetBody<Product>().ProductID.ToString());
                    }

                    

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

            Console.ReadLine();

 

特別注意：消費掉Product或Customer里的消息后，AllMessage里的消息不受影響。