Azure Services Bus(服務總線)中的工作流(workflow)

作者: 代震軍來源: 博客園發布時間: 2008-12-28 13:50 閱讀: 1686 次推薦: 0 原文鏈接 [收藏]

摘要：在Azure Services Platform平臺中.NET Services支持工作流服務的。這篇文章通過一個Demo介紹了WorkflowService。

在Azure Services Platform上對于工作流服務的支持，一直是我很感興趣的內容。當然也是疑問比較多的領域。鑒于這方面的資料太少，所以今天就從AzureServicesKit中的一個DEMO出發，來大概了解一下這方面相關內容。

注：今天的示例位于AzureServicesKit安裝目錄\Labs\Ex02-RoutingWithXPath\end文件夾。

(編輯注：是AzureServicesKit\Labs\IntroWorkflowService\Ex02-RoutingWithXPath\end文件夾)

該示例場景展示的是一個定單(order)流程，如下圖：

注：圖中的兩個服務可能布置在1臺或N臺機器上。

在上圖中，我們看出在當前場景中存在兩個服務，即：

BillService（即定單生成）。 ShipOrderService(定單處理：包括處理定單相關信息等)

其中的BillService的代碼如下：

[ServiceContract(Name = "Billing",
    Namespace = "http://Microsoft.ServicePlatformLabs")]
public class BillingService
{
    [OperationContract(Name = "Invoice", IsOneWay = true)]
    public void Invoice(string orderId, string total)
    {
        Console.WriteLine("Invoice for Order {0} ({1}) generated",
            orderId, total);
    }
}

注：上面的ServiceContract屬性Name="Billing"和OperationContract屬性Name = "Invoice"會以鏈接方式綁定到工作流CloudServiceBusSend活動(activity)的Action屬性上，即：

http://Microsoft.ServicePlatformLabs/Billing/Invoice

ShippingService的代碼如下：

[ServiceContract(Name = "Shipping",
    Namespace = "http://Microsoft.ServicePlatformLabs")]
public class ShippingService
{
    [OperationContract(Name = "ProcessOrder", IsOneWay = true)]
    public void ProcessOrder(string orderId)
    {
        Console.WriteLine("Processing Shipping information for Order {0}",
            orderId);
    }
}

同理，ShippingService會以鏈接方式綁定到工作流CloudServiceBusSend活動(activity)的Action屬性上，即：

http://Microsoft.ServicePlatformLabs/Shipping/ProcessOrder

而這兩個服務都會被暴露到ServiceBus中以便讓用戶進行訪問操作，從而完成一個客戶下訂單的完整流程（CreateOrder）

而如何對這兩個服務進行安排組裝，就是通過WorkFlow的進行的。如果有開發過工作流經驗的開發者應該會很容易理解這個概念。不過這里還是要解釋一下，就是在云中運行的工作流與我們平時所了解的工作流（主要是在Activities方面）還是有些差異的，當然這并不意味著云中的工作流要更難于理解，恰恰相反，就目前而言，還是很容易的，下面是其在VS中的設計器中的截圖：

有關這幾個新添加的工作流activity，可能就要幾個篇幅來介紹和說明，因為今天的目的不在于此，所以就先略過這塊內容了。

另外就是目前在云中只支持CloudSequentialWorkFlow,如下圖：

目前還不支持狀態機工作流，但并不確保將來就不會出現。當然將來會不會出現workflow4中的flowchart型工作流就更不好說了。

假設我們最終要實現的工作流程如下：

1.接受客戶端post過來的請求，并獲取其中指定的節點信息，本DEMO中節點路徑為：

<root>
  <order>
    <total>節點值</total>
  </order>
</root>

2.通過對該節點值進行比較判斷，當值大于1000時則將工作流程轉入一個CloudDelay活動中，以進行延時操作（設置CloudDelay活動的TimeOut屬性）。當值小于或等于1000時則順序執行上面所介紹的兩個服務（BillingService，ShippingService）

將上面的流程中工作流進行表示（創建）的結果如下圖：

好了，現在我們有了工作流和服務，接下來就是通過編碼將服務運行起來以及將工作流發布到Azure上了。下面是服務的創建運行代碼：

internal static void Main()
{
    var billingServiceHost = new ServiceHost(typeof(BillingService));
    Console.WriteLine("BillingService hosted at:");
    Console.WriteLine("\t" + billingServiceHost.Description.Endpoints[0].Address.Uri);
    billingServiceHost.Open();

    var shippingServiceHost = new ServiceHost(typeof(ShippingService));
    Console.WriteLine("ShippingService hosted at:");
    Console.WriteLine("\t" + shippingServiceHost.Description.Endpoints[0].Address.Uri);
    shippingServiceHost.Open();

    Console.WriteLine();
    Console.WriteLine("Press [Enter] to exit");
    Console.ReadLine();

    billingServiceHost.Close();
    shippingServiceHost.Close();
}

上面代碼中的Address.Uri屬性是在app.config中進行配置的：

代碼在這里展開

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
  <system.serviceModel>
    <bindings>
      <netEventRelayBinding>
        <binding name="default" />
      </netEventRelayBinding>
    </bindings>

    <services>

      <service name="Services.ShippingService">
        <host>
          <baseAddresses>
            <add baseAddress="sb://servicebus.windows.net/services/[ENTER YOUR SERVICEBUS USERNAME HERE]"/>
          </baseAddresses>
        </host>

        <endpoint address="Shipping"
                  contract="Services.ShippingService"
                  binding="netEventRelayBinding"
                  behaviorConfiguration="transportClientBehavior0"
                  bindingConfiguration="default" />

      </service>

      <service name="Services.BillingService">
        <host>
          <baseAddresses>
            <add baseAddress="sb://servicebus.windows.net/services/[ENTER YOUR SERVICEBUS USERNAME HERE]"/>
          </baseAddresses>
        </host>

        <endpoint address="Billing"
                  contract="Services.BillingService"
                  binding="netEventRelayBinding"
                  behaviorConfiguration="transportClientBehavior0"
                  bindingConfiguration="default" />

      </service>
    </services>

    <behaviors>
      <endpointBehaviors>
        <behavior name="transportClientBehavior0">
          <!--
            Specify
              - CardSpace (default)
              - UserNamePassword
              - X509Certificate
              - AutomaticRenewal (tgt)
              - FederationViaCardSpace
            for credentialType.
          -->
          <transportClientEndpointBehavior credentialType="UserNamePassword">
            <clientCredentials>
              <userNamePassword userName="[ENTER YOUR SERVICEBUS USERNAME HERE]"
                                password="[ENTER YOUR SERVICEBUS PASSWORD HERE]" />
            </clientCredentials>
          </transportClientEndpointBehavior>
        </behavior>
      </endpointBehaviors>
    </behaviors>
  </system.serviceModel>

</configuration>

注:config文件中的[ENTER YOUR SERVICEBUS USERNAME HERE]內容就是我們在AzureService平臺上創建的solution名稱（這部分內容參見這篇文章），這里我們繼續使用上一篇文章中創建的那個項目名稱MSF_DataSyncExample,而PASSWORD就是我們在創建MSF_DataSyncExample之后所設置的口令。

上面僅是完成了服務的本地化配置。而要把工作流（文件內容）發布到AzureService上我們還需要在工作流文件的可視模式下的空白區域單擊鼠標右鍵，從彈出菜單中選擇“Deploy WordFlow”,這里系統會彈出一個窗口，如下：

我們只需要將剛才在config文件中輸入的solution名稱和口令在這里敲進去并單擊Deploy按鈕即可。

這樣系統就會在AzureService上為我們創建這樣一個發布了的工作流（注：您也可以通過下面的地址http://workflow.ex.azure.microsoft.com/login.aspx?name=[solutionname]來訪問AzureService并在系統向導的提示下完成工作流的發布）。

工作流發布之后，我們還需要為當前的工作流創建一個實例，以便于讓客戶端訪問該工作流服務時持有它，進而完成工作流程。而創建工作流實例的工作是在Azure平臺上完成的，請在IE地址欄中敲入地址：https://workflow.ex.azure.microsoft.com/WorkflowManagement.aspx，如下圖：

我們在上圖中選擇了剛才發布(Deploy)的工作流，然后點擊“Manage Instances”按鈕后，系統顯示如下：

我們點擊：“CreateInstance”按鈕后，如下圖：

我們看到，系統以 “name_date/time stamp”方式為我們命名了一個“實例名稱”，這主要是為了避免出現同名的情況，因為實例在系統中必須唯一。

接著我們點擊"next"按鈕，系統接下來會提示我們進行身份驗證，如下圖：

這樣系統就會創建該實例并返回到當前工作流的實例管理列表：

當我們把工作流實例創建完成后，就把將當前工作流運行（通過選中相應的實例并單擊頁面下方的“Start”按鈕，另外可以點擊Details按鈕來獲得實例ID，下面會介紹）起來。

這樣在Azure平臺上，我們的工作流就算是配置運行起來了，那么接下來，我們來看看到如何讓客戶端訪問發布的工作流的。

首先是客戶端的配置文件

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
  <appSettings>
    <add key="UserName" value="[ENTER YOUR WORKFLOW USERNAME HERE]"/>
    <add key="Password" value="[ENTER YOUR WORKFLOW PASSWORD HERE]"/>
    <add key="InstanceId" value="[ENTER THE WORKFLOW INSTANCE ID]"/>
  </appSettings>
</configuration>

從上面節點信息可以看出，我們要將配置在service端的相應節點信息（包括solution名稱和口令）配置到當前config節點上。當然這里還多了一個節點就是“InstanceId”，該實例ID就是我們剛才在Azure平臺上創建的工作流實例的ID,我們可以通過單擊相應的實例來獲得該內容，如下：

這樣就完成了在client端的config配置，下面就是客戶端訪問的工作流的代碼了。

代碼在這里展開

internal class Program
{
    private const string workflowTypeName = "CloudOrderWorkflow2";
    private const string activityName = "ReceiveOrder";
    
    internal static void Main()
    {
        Console.Title = "Client";

        var userName = ConfigurationManager.AppSettings["UserName"].ToString();
        var password = ConfigurationManager.AppSettings["Password"].ToString();
        var instanceId = ConfigurationManager.AppSettings["InstanceId"].ToString();

        Console.WriteLine("Press [enter] to send the order to the cloud workflow");
        Console.ReadLine();

        var request = CreateMessage();

        try
        {
            CallReceiveActivity(userName, password, workflowTypeName, instanceId, activityName, request);
        }
        catch (WebException ex)
        {
            Console.WriteLine("Error handled:");
            using (var sr = new StreamReader(ex.Response.GetResponseStream()))
            {
                Console.WriteLine(sr.ReadToEnd());
            }
        }

        Console.WriteLine("Press [enter] to exit");
        Console.ReadLine();
    }

    private static string CreateMessage()
    {
        return "<root><order><total>1400</total></order></root>";
    }

    internal static void CallReceiveActivity
        (string userName, string password, string workflowName,
        string instanceId, string activityName, string message)
    {
        var url = String.Format(CultureInfo.InvariantCulture,
                   "http://{0}/workflowsHttp/{1}/workflows/{2}/instances/{3}/{4}",
                   WorkflowClientConfig.WorkflowHostName, userName, workflowName, instanceId, activityName);
        
        var token = GetWorkflowGrantingToken(userName, password);

        using (var httpRequest = new WebClient())
        {
            httpRequest.Headers.Add(HttpRequestHeader.ContentType, "text/xml");
            httpRequest.Headers.Add("X-MS-Identity-Token", token);
            var response = httpRequest.UploadString(url, message);

            Console.WriteLine("Response by the server: {0}", response);
        }
    }

    internal static string GetWorkflowGrantingToken(string userName, string password)
    {
        string tokenUrl = string.Format(CultureInfo.InvariantCulture,
                                        "https://{0}/issuetoken.aspx?u={1}&p={2}",
                                        WorkflowClientConfig.StsHostName,
                                        userName,
                                        password);

        Console.WriteLine("Requesting authentication token...");
        HttpWebRequest tokenRequest = (HttpWebRequest)WebRequest.Create(tokenUrl);
        HttpWebResponse tokenResponse = (HttpWebResponse)tokenRequest.GetResponse();

        byte[] tokenBody = new byte[500];
        int tokenBodyLength = tokenResponse.GetResponseStream().Read(tokenBody, 0, 500);

        var authenticationToken = Encoding.UTF8.GetString(tokenBody, 0, tokenBodyLength);
        Console.WriteLine("Authentication Token: {0}", authenticationToken);

        return authenticationToken;
    }
}

現在我們在本地分別打開兩個VS2008,一個將service項目做為啟動項目中，一個以Client作為啟動項目，分別在相應的項目上擊鼠標右鍵，在彈出菜單上選“調試”--》“啟動新實例”。先運行service端如下：

然后是按上面所說方式啟動Client端：

這里，我們在切換回服務端運行界面，如下

到這里，我們只是測試了工作流中的ELSE分支（代碼中Total為800），還未測試當Total>1000的分支。下面是修改Total值之后的測試流程：

1.首先重復上面在AZURE平臺上創建工作流實例的步驟，以便生成新的實例ID供客戶端config文件更新節點配置。

2.將client的運行代碼從800改為1400,然后順序啟動上面的service,client應用，這時我們還是會看到之前運行的service端和client端的運行界面，只是這次服務端在客戶端提提交請求之后，并未馬上返回打印結果，而是運行了cloudDelay活動，這時如果我們訪問azure平臺上的相關工作流頁面時，會看到下面的結果 azure_run_flow.gif 即當前工作流實例正在運行，我們需要等待1分鐘之后，再刷新該頁面時，會看到當前實例已運行完成： azure_workflow_complate.gif

好了，到現在為止，整個開發和運行測試流程就介紹的差不多了。

因為手頭的資料不多，而我還有的一些疑問還是沒有最終得到解答。下面我將它們羅列出來，如果大家感興趣或有這方面的經驗，不妨與我聯系或在回復中進行討論。

1.例子中的SERVICE端如果能夠被布署在多臺服務器上，當client端post請求時，當前會通過哪臺機器上的服務來處理？我猜測的一種可能應該是service bus會通過一些網絡路由算法（如最短路徑算法）來分配相應的服務機來處理相應請求。但如果是這樣，是否還應該包括負載均衡方面的考慮呢？必定每臺機器的處理能力有限，任務多了就要排隊，如果一味還是”最短路徑“的話，會讓service bus中的某一個服務結點不堪負。

2.目前的測試可以讓azure平臺上的某一工作流在某一條件下被觸發執行（上傳中的total<=1000）.但工作流的持久化就在azure上嗎？這樣的話，如果azure平臺出現問題，正在運行的實例可能會將執行在一半的工作流“回滾”，以免出現數據被臟讀的情況嗎？另外如果企業想將被執行失敗的工作流按“自己的方式”進行處理又應該如何去做呢？

3.在biztalk中其扮演的角色是ESB（企業服務總線），而Azure Service Bus目前而言應該是一個ISB(Internet 服務總線)。如果與ISG是ESB的一種實現方式的話，那在ESB中的消息類型與ISB中的消息類型又是否為一脈相承呢，在SDK中我看到了這樣一段代碼（出現在了AzureServicesKit\Labs\IntroServiceBus\Ex04-RESTSample中）：

Message response = StreamMessageHelper.CreateMessage(OperationContext.Current.IncomingMessageVersion, "GETRESPONSE", this.WriteImage);

而該Message類型為abstract類型（位置System.ServiceModel.Channels名空間），這個類型又與在BIZTALK Server上配置于數據庫中的message表中的消息數據是怎樣一個關系呢。以前曾在網上看到有篇文章說azure會將shartpoint逐步在云中加以實現，如果是這樣的話那biztalk是否將來會與AZURE平臺產生某種更深層次的關聯嗎？

疑問之后還是疑問，看來在研究Azure平臺的過程中還有很長的路要走，走一步看一步吧!!!