Azure service bus Topic基本用法

spell

　　我们在升级一个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创建，也可以通过代码创建：

　　通过代码创建主题：





1 // Create the topic if it does not exist already.
2             string connectionString =
3                 CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
4
5             var namespaceManager =
6                 NamespaceManager.CreateFromConnectionString(connectionString);
7
8             if (!namespaceManager.TopicExists("TestTopic"))
9             {
10
11                 //默认的createtopic方法
12                  namespaceManager.CreateTopic("TestTopic");
13
14
15                 // 通过TopicDescription构建一个重载
16                 //TopicDescription td = new TopicDescription("TestTopicCustomer");
17                 //td.MaxSizeInMegabytes = 5120;
18                 //td.DefaultMessageTimeToLive = new TimeSpan(0, 1, 0);
19                 //if (!namespaceManager.TopicExists("TestTopicCustomer"))
20                 //{
21                 //    namespaceManager.CreateTopic(td);
22                 //}
23
24             }
View Code　　4.创建订阅：
　　下面的代码演示创建了三个订阅，其中Product和Customer订阅增加了一个SqlFilter，即如果发送消息时，
　　message.Properties["Entity"] = "Product"，则消息会由Product订阅处理；
　　message.Properties["Entity"] = "Customer"，则消息会由Customer订阅处理；





1  string connectionString =
2     CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
3
4             var namespaceManager =
5                 NamespaceManager.CreateFromConnectionString(connectionString);
6
7             if (!namespaceManager.SubscriptionExists("TestTopic", "AllMessages"))
8             {
9                 namespaceManager.CreateSubscription("TestTopic", "AllMessages");
10             }
11
12             Console.WriteLine("AllMessages done");
13
14
15             // Create a "ProductMessagesFilter" filtered subscription.
16             SqlFilter ProductMessagesFilter =
17                new SqlFilter("Entity='Product'");
18
19             namespaceManager.CreateSubscription("TestTopic",
20                "Product",
21                ProductMessagesFilter);
22
23             Console.WriteLine("Product done");
24
25             // Create a "CustomerMessagesFilter" filtered subscription.
26             SqlFilter CustomerMessagesFilter =
27                new SqlFilter("Entity='Customer'");
28
29             namespaceManager.CreateSubscription("TestTopic",
30                "Customer",
31                CustomerMessagesFilter);
32
33             Console.WriteLine("Customers done");
View Code　　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条消息。





1  string connectionString =
2     CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
3
4             TopicClient Client =
5                 TopicClient.CreateFromConnectionString(connectionString, "TestTopic");
6
7            // Client.Send(new BrokeredMessage());
8
9
10             //Add 5 Products
11             var product = new Model.Product();
12
13             for (int i = 0; i < 5; i++)
14             {
15
16                 product.ProductID ="ProductID"+ i.ToString();
17
18                 // Create message, passing a Product  for the body.
19                 BrokeredMessage message = new BrokeredMessage(product);
20
21                 // Set additional custom app-specific property.
22                 message.Properties["Entity"] = "Product";
23
24                 //ScheduledMessage
25                 //message.ScheduledEnqueueTimeUtc = DateTime.Now.AddMinutes(1);
26
27                 // Send message to the topic.
28                 Client.Send(message);
29             }
30
31
32             //Add 3 Customers
33             var customer = new Model.Customer();
34
35             for (int i = 0; i < 3; i++)
36             {
37
38                 customer.CustomerID = "CustomerID"+i.ToString();
39
40                 // Create message, passing a Customer message for the body.
41                 BrokeredMessage message = new BrokeredMessage(customer);
42
43                 // Set additional custom app-specific property.
44                 message.Properties["Entity"] = "Customer";
45
46                 // Send message to the topic.
47                 Client.Send(message);
48             }
View Code　　发送消息后的效果：

　　 创建定时消息：
　　设置消息的如下属性，则消息在发送成功后延时1分钟后才会出现在上图中，也才可以被订阅消费。
　　message.ScheduledEnqueueTimeUtc = DateTime.Now.AddMinutes(1);
　　6.消费消息：
　　不同的订阅可以取不同订阅里的消息进行处理，处理完成后，标记message.Complete()即可从订阅里清除消息。





1 string connectionString =
2     CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
3
4             //SubscriptionClient Client =
5             //    SubscriptionClient.CreateFromConnectionString
6             //            (connectionString, "TestTopic", "Customer");
7
8             //SubscriptionClient Client =
9             //   SubscriptionClient.CreateFromConnectionString
10             //           (connectionString, "TestTopic", "Product");
11
12             SubscriptionClient Client =
13               SubscriptionClient.CreateFromConnectionString
14                       (connectionString, "TestTopic", "AllMessages");
15
16             // Configure the callback options.
17             OnMessageOptions options = new OnMessageOptions();
18             options.AutoComplete = false;
19             options.AutoRenewTimeout = TimeSpan.FromMinutes(1);
20
21             Client.OnMessage((message) =>
22             {
23                 try
24                 {
25                     // Process message from subscription.
26                     Console.WriteLine("Messages");
27               
28                     Console.WriteLine("MessageID: " + message.MessageId);
29                     Console.WriteLine("Message EntityType=: " +
30                         message.Properties["Entity"]);
31
32
33                     if(message.Properties["Entity"].ToString()=="Customer")
34                     {
35                         Console.WriteLine("CustomerID=" + message.GetBody<Customer>().CustomerID.ToString());
36                     }
37                     if (message.Properties["Entity"].ToString() == "Product")
38                     {
39                         Console.WriteLine("ProductID=" + message.GetBody<Product>().ProductID.ToString());
40                     }
41
42                     
43
44                     // Remove message from subscription.
45                     message.Complete();
46                 }
47                 catch (Exception)
48                 {
49                     // Indicates a problem, unlock message in subscription.
50                     message.Abandon();
51                 }
52             }, options);
53
54             Console.ReadLine();
View Code　　特别注意：消费掉Product或Customer里的消息后，AllMessage里的消息不受影响。