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创建，也可以通过代码创建：
通过代码创建主题：
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订阅处理；
1string 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条消息。
1string 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 Productfor 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里的消息不受影响。
页:
[1]