【Microsoft Azure学习之旅】消息服务Service Bus的学习笔记及Demo示例

　　今年项目组做的是Cloud产品，有幸接触到了云计算的知识，也了解并使用了当今流行的云计算平台Amazon AWS与Microsoft Azure。我们的产品最初只部署在AWS平台上，现在产品决定同时支持Azure，所以有幸学习下Azure，并在查看文档资料以及写Demo过程中发现了其中的一些不同。虽然AWS与Azure是两款旗鼓相当的竞争产品，但是还是有很多区别。

　　本文主要是自己学习Service Bus中的学习笔记，自己有些结论也都跟微软技术支持确认过。个人观点，抛砖引玉：-）

　　消息服务对于云计算产品是很重要的。例如SQS，我们可以使用持久化队列来储存消息，这样同一产品的不同Module在通信过程中利用队列存储消息，可避免消息的丢失（因为消息会被分布式冗余的存储，属于队列的内部实现）。针对队列服务，AWS提供了SQS，Azure则提供了Service Bus（当然除了Service Bus，微软还提供了Storage部分中的Queue功能，但个人认为这个Queue偏重消息存储，而Service Bus中的Queue偏重通信，具体区别可参照此文章http://msdn.microsoft.com/library/azure/hh767287）。例外AWS里面，SNS可以和SQS一起使用。SNS中就提供了对SQS的支持，如果有消息产生需要通知Module，SNS可以直接将消息存储在SQS中。

　　而微软的Azure提供的Service Bus功能，主要有五部分子功能组成：

　　1. Queue 队列

　　　 提供的功能与Amazon SQS类似

　　2. Topic/Subscription 主题/订阅

　　　 更高级的Queue，类似与一个虚拟队列，可接收发送到主题的消息副本，从订阅接收消息的方式与队列接收消息的方式相同。

　　3. Relay 中继

　　　 创建Hybrid Application时使用，我没有使用过此功能。

　　4. Notification Hub 通知中心

　　　 主要是向Device移动设备推送通知，例如注册一个APP在通知中心，通知中心可向运行此APP的所有设备发送推动通知，支持几乎所有的手机平台。

　　5. Event Hub 事件中心

　　可用于存储产品运行时产生或收集的大量事件。

　　这里我们的产品主要使用的将是第二项功能，主题/订阅。这样我们产品中的一个Module只需将消息发送至主题中，凡是订阅了该主题的订阅（队列）都是有一个该消息的副本。这样我们的另一个Module只需关注订阅中的消息，取出消息，处理消息，达成通信的目的。

　　在Azure中，要使用主题/订阅，首先要在Service Bus中新建一个Namespace，然后新建主题与订阅即可。

　　值得注意的是，之前Service Bus提供的认证方式都是ACS，前一段时间都改为了SAS（共享访问签名）。而这一改动需要SDK的支持，据我所知，Azure Java SDK上周才刚刚在新版本中支持了这一认证，可谓效率之低。具体详情可参见我上一篇文章：【Microsoft Azure学习之旅】Azure Java SDK - Service Bus的认证问题http://www.cnblogs.com/KevinSong/p/4146811.html。

　　下面的代码是我根据微软文档写的一个Demo。使用Java SDK实现，作用是向主题发送消息，然后从订阅中取出消息。

　　1. 创建一个主题，并向主题发送消息

//create a topic in name space
public void Create(TopicInfo topic){
    System.out.println("=>CreateTopic");
    try{
        if(!IsAlreadyExist(topic)){
            CreateTopicResult result = service.createTopic(topic);
            System.out.println("Create Topic Result: " + result.toString());
        }
        else{
            System.out.println("This Topic already exist, doesn't need to create");
        }
    }
    catch (ServiceException e){
        System.out.println("ServiceException encountered: " + e.getMessage());
    }
    System.out.println("<=CreateTopic");
}

//send message to a topic
public void SendMessage(TopicInfo topic){
    System.out.println("=>SendMessage");
    try{
        BrokeredMessage msg = new BrokeredMessage("testMsg");
        
        //set one property to this msg
        msg.setProperty("testProperty", "kevin01");
        
        service.sendTopicMessage(topic.getPath(), msg);
    }
    catch (ServiceException e){
        System.out.println("ServiceException encountered: " + e.getMessage());
    }
    System.out.println("<=SendMessage");
}

　　2. 创建一个关注该主题的订阅，并从订阅中取出消息

//create a subscription in name space
public void Create(TopicInfo topic, SubscriptionInfo sub){
       System. out.println("=>CreateSubscription" );
       try{
          System. out.println("Topic Info: " + topic .getPath().toString());
         
          if(!IsAlreadyExist(topic , sub )){
             CreateSubscriptionResult result = service.createSubscription(topic .getPath(), sub);
              //and we can create a rule for this topic/subscription
              //RuleInfo rule = new RuleInfo();
             System. out.println("Create Subscription Result: " + result.toString());
          }
          else{
             System. out.println("This subscription already exist. No need to create");
          }
      }
       catch (ServiceException e ){
          System. out.println("ServiceException encountered: " + e.getMessage());
      }
      System. out.println("<=CreateSubscription" );
}


//receive message from subscription
public void ReceiveMessage(TopicInfo topic, SubscriptionInfo sub){
      System. out.println("=>ReceiveMessage" );
      ReceiveMessageOptions opts = ReceiveMessageOptions. DEFAULT;
       opts.setReceiveMode(ReceiveMode. PEEK_LOCK);
       try{
             ReceiveSubscriptionMessageResult result = service.receiveSubscriptionMessage(topic .getPath(), sub.getName(), opts);
             
             BrokeredMessage msg = result.getValue();
             
              if(msg != null && msg.getMessageId() != null){
                    //print the message info
                   System. out.println("Body: " + msg .toString());
                   System. out.println("Message ID: " + msg.getMessageId());
                   
                   System. out.println("Custom Property Value: " + msg.getProperty("testProperty" ));
                   
                    //delete this message
                    service.deleteMessage( msg);
             }
              else{
                   System. out.println("There's no message in this subscription. Topic: " + topic.getPath() + ", Subscription: " + sub .getName());
             }
      }
       catch (ServiceException e ){
          System. out.println("ServiceException encountered: " + e.getMessage());
      }
      System. out.println("<=ReceiveMessage" );
}

　　3. Main函数

//test topic/subscription
public static void main(String[] args){
    //create a topic, and send a message to this topic
    ServiceBusTopicHandler topicHandler = new ServiceBusTopicHandler();
    TopicInfo topic = new TopicInfo("testtopic");
    topicHandler.Create(topic);
    topicHandler.SendMessage(topic);
    
    //create a subscription about this topic, and receive message from this subscription
    ServiceBusSubscriptionHandler subHandler = new ServiceBusSubscriptionHandler();
    SubscriptionInfo sub = new SubscriptionInfo("testsubscription");
    subHandler.Create(topic, sub);
    
    while(true){
        subHandler.ReceiveMessage(topic, sub);
    }
}

　　得益于微软提供的文档与SDK，代码实现很简单。但是有个地方值得注意的是，

　　如何从订阅/队列中取出消息？

　　这涉及到Poll和Push技术的区别。

　　最简单的是我们可以使用Poll技术，就是例如在while(true)循环中一遍遍去查询，如果有新消息就取出并处理，但是这样会大大影响性能。在AWS中，我们可以利用SNS技术中的一项，注册http endpoint到SNS，如果有新消息来了，可以通过SNS调用我们自己的Web API来通知我们要去处理新消息，这样就是Push的效果。但是在Azure中，有类似SNS的功能吗？没有。

　　这样SNS的技术类似于Push技术，如果有新消息，它会主动调用你的API通知你。而非Poll那样低效率的一遍遍查询。但是微软Azure在去年提出了一项Long Polling的新技术，它的名字叫做Event-Driven Message Programing Model，事件驱动消息编程模型，该技术的发布信息可参考http://msdn.microsoft.com/en-us/library/azure/dn198643.aspx。这其实本质是一项Long Polling长轮询技术。

　　在Azure .Net SDK中，你可以使用OnMessage方法，来注册一个回调Call back函数，如果有新消息来到，你的Call back会被调用到，来实现对消息的处理。本质是有不同的子线程，在执行long polling查询。这跟Push技术还是有很大的区别。

　　但现在问题是，Event-Driven Message Programming Model仅仅在.Net SDK中提供，在其他语言的Azure SDK版本中并不提供此项功能。曾经今年有人在MSDN问过这个问题，为什么不同的SDK区别对待，微软的回复请参照https://social.msdn.microsoft.com/forums/windows/en-us/7d0d4733-a696-4b72-927c-004caae029f7/event-driven-consumer-model-not-available-for-microsoft-azure-java-sdk?forum=windowsazuredevelopment。

　　由此可见，微软对于.NET SDK的重视，而对于类似Java的不重视啊，毕竟是自家的语言。.Net SDK已经目前发布了数百个Release，而Java SDK还仅仅是v0.7.0版本。。。当然通过询问微软，Java SDK在将来有可能也支持此项功能（Event-Driven Message），但具体什么时间会支持一切未可而知。

　　

　　无论是AWS，还是Azure，都有完整详细的文档提供，这里只是我在学习过程的一些总结。我的代码示例（Java实现，采用Azure Java SDK）也附上如下。

　　Service Bus Queue/Subject/Subscription Demo下载地址：http://files.cnblogs.com/KevinSong/testServiceBusQueue.zip

　　有任何问题，欢迎讨论。小白一个，继续学习：-）

参照文档：

1. Service Bus队列，主题和订阅

　 http://msdn.microsoft.com/library/azure/hh367516.aspx

2. 介绍Event-Driven Message Programming Model

　 http://fabriccontroller.net/blog/posts/introducing-the-event-driven-message-programming-model-for-the-windows-azure-service-bus/

3. AWS与Azure区别

　 http://azure.microsoft.com/en-us/campaigns/azure-vs-aws/

Best Regards

Kevin Song

2014年12月19日