WF4 中的动态发送活动

Roman Kiss

4.43/5 (6投票s)

2010 年 5 月 29 日

CPOL

14分钟阅读

64482

1003

本文描述了 .Net WF4 技术中自定义动态发送活动的 **设计、实现和使用**。

下载源代码

特点

发送活动的服务绑定范围
动态更改绑定
动态插入消息头
内置的 QueueMessageSend 活动
.Net 4 技术

引言

本文将比我以前的文章要短。它关注的是 WF4 范例中 Send 活动的可扩展性。Send 活动代表了在特定消息交换模式 (MEP) 中消耗服务的连接原语。基于设置的属性，WCF 通道会创建一个消息并将其发送到工作流域之外。下面的屏幕截图显示了其属性网格

Send 活动的使用非常直接，基本上消息契约和 AddressBindingContract (ABC) 属性在设计时设置并在运行时执行。在 Send 活动中，运行时灵活性非常有限，例如，我们可以使用通用的非类型化消息契约来创建通用代理（有关此消息调解的更多详细信息，请参阅我之前的文章此处），或者我们可以选择配置文件中的客户端配置等。

如上图所示，Endpoint 属性必须在设计时声明，运行时无法声明性地动态选择绑定，例如基于规则存储库。Send 活动的另一个运行时要求是在运行时根据存储库资源向传出的消息添加应用程序头。

换句话说，我们需要一个动态发送活动来控制其 ABC 属性。下图显示了围绕 Send 活动的 BindingScope 的自定义活动示例

在此自定义活动中，Binding 是一个具有 dynamic 类型的 ExpressionTextBox (ETB) 属性，因此我们可以使用 XElement、String 或 Binding 类来显式或隐式地通过变量声明绑定。根据上图，Send 活动在其 xaml 资源中拥有自己的绑定节，而不是在配置文件中。

为了隐藏绑定节中声明的一些绑定复杂性（请参阅上图中的 netMsmqBinding），我们可以为 Send 活动预构建自定义绑定范围，以用于特定绑定。下图显示了一个用于 MSMQ 消息的自定义 SendQueueMessage 活动

如您所见，上述动态发送活动将在运行时创建一个 netMsmqBinding 绑定，并将其分配给 Send.Endpoint 属性。

BindingScope 代表 Send 活动周围的绑定包装器。以下屏幕截图显示了 SendQueueMessage 绑定的范围

请注意，Send 活动不需要任何更改，因此我们可以将此范围用于在设计的工作流中声明的 Send 活动，只需将 Send 活动拖放到范围区域，绑定范围就会覆盖其硬编码的（组合框选择的）Endpoint 属性。

好的，让我们开始讨论动态发送活动的概念和设计。

我假设您有一些 WF4 Send 活动的工作经验或理解其功能。

概念与设计

动态发送活动的概念和设计基于 System.Activities.NativeActivity 的功能。派生自 NativeActivity 的自定义活动将拥有对工作流运行时引擎模型的所有访问权限。以下代码片段显示了此基本概念

[ContentProperty("Body")]
[DesignTimeVisible(false)]
public sealed class BindingScope : NativeActivity 
{
  // Inputs (Binding, Headers)
  
  // Body (Send Activity)
  
  // Targeting (runtime updating a Body)
}

NativeActivityContext 在活动执行期间传递，可用于使用 ScheduleActivity 方法调度其他活动（在我们设计中是 Send 活动）。这是 WF4 模型的一个非常强大的功能，允许我们在活动执行期间对其进行预处理。

以下代码片段显示了此重写方法，其中 this.Body 是我们的范围性 Send 活动。在将此活动传递执行之前，我们可以更新其绑定并注册一个回调函数来注入消息头

protected override void Execute(NativeActivityContext context)
{   
    if (this.Headers != null)
    {
        // inject headers
    }

    if (this.Binding != null)
    {
        dynamic dynBinding = this.Binding.Get(context);

        // update binding
    }

    context.ScheduleActivity(this.Body);
}

第二个重要的重写方法是 CacheMetadata。此方法会验证我们的 Send 活动并设置一个虚拟 Endpoint

protected override void CacheMetadata(NativeActivityMetadata metadata)
{
  if (this.Body == null)
    metadata.AddValidationError("Missing a Send Activity in the scope");

  if (this.Body != null && this.Body.Endpoint == null)
    this.Body.Endpoint = new Endpoint() 
      { 
        Binding = new BasicHttpBinding(), 
        AddressUri = new Uri("https:///Fake") 
      };

  metadata.AddChild(Body);
}

正如您所见，动态发送活动的概念非常直接，它利用了 WF4 范例的可扩展性。在本文中，我仅演示了绑定范围和注入消息头，但它可用于任何属性，例如 KnownTypes。范围活动可以有一个简单的设计器，也可以有更复杂的设计器，例如访问存储库以获取绑定模板、消息头等。

让我们描述一下本文解决方案中包含的所有自定义动态发送活动。

SendScope

SendScope 是一个基本的自定义活动，用于在具有最小设计器的 Send 活动周围进行范围限定。

此范围有两个属性，第一个用于声明绑定，第二个用于消息头。如我之前提到的，ETB 属性被定义为动态类型，因此绑定可以像下图所示那样声明

以及以下图片显示了 Headers 属性的示例

此自定义活动的使用非常直接，就像 Send 活动一样。此外，绑定和消息头可以通过 xaml 动态声明。

BindingScope

BindingScope 是一个自定义活动，与 SendScope 活动类似，带有用于绑定节的设计器。下图显示了此设计

Binding 是一个多行文本的 ETB 属性，允许我们直接从设计器表面编辑 xml 文本。绑定具有与配置文件中绑定节相同的 xml 格式文本。请注意，本文发布了一个没有编辑器对话框、无法访问绑定模板存储库等的简单设计器。

SendQueueMessage

SendQueueMessage 是一个自定义范围活动，围绕 Send 活动，用于特定绑定，如 NetMsmqBinding。下图显示了其设计器和属性

SendQueueMessage 允许发送具有自定义消息头的事务性消息到位于特定计算机上的私有队列。消息可以通过 TimeToLive 和 DeadLetterQueue 属性进行控制。此自定义活动的使用非常强大。让我们在以下段落中深入探讨此消息交换模式。

MSMQ 4.0 是一个伟大的微软技术。它是免费的（作为操作系统的一部分），并且是用于本地或远程队列之间快速、可靠或事务性消息传递的非常可靠的消息传递机制。WCF 具有内置的 net.msmq 通道，用于抽象访问 MSMQ 范例并虚拟化服务与其使用者之间的连接。在 WCF/WF 技术之前，我很久以前就构建了这个抽象和虚拟化，用于自定义 Remoting 通道（请参阅我发布的关于 .Net 1.0 的文章此处），并在多个项目中用于将业务层与物理通信通道进行封装。此概念允许非常平滑地将遗留应用程序迁移到 WCF/WF 技术。

这个故事的结局是将应用程序分解成小的业务动作（流程），并使用 MSMQ 技术组合它们，从而能够以松耦合的事件驱动的方式管理和扩展我们的应用程序。

下图显示了一个典型的场景，用于将过程分解为管理器和工作进程。管理器遍历一组规则，并通过 MSMQ 为工作进程生成消息

Worker Service 可以通过设置合适的 خدمة 节流配置来控制，这取决于托管能力等。SendQueueMessage 活动可用于向工作进程发送事务性消息

上图显示了一个在 Search 队列中的消息（带有自定义消息头），必须在 30 秒内收到。未指定 DeadLetterQueue（DLQ），因此未送达的消息将丢失。

让我们创建一个更现实的场景，例如，工作进程服务因某种原因失败，其工作必须再次执行。我们可以为这个需求提供更多解决方案，例如创建一个自定义的 Retry 活动等，但是，通过使用 MSMQ 4.0 的功能，只需重新配置 SendQueueMessage 即可声明性地实现。这个神奇的功能基于 DeadLetterQueue，无需监听器/接收器。

下图显示了这个场景。

如您所见，我们有一个额外的队列（WaitQueue），但没有监听器。消息重试技巧基于在 TimeToLive 时间过期后将消息从等待队列移动到死信队列。下图显示了用于工作进程服务的 SendQueueMessage 属性

现在，我拥有了使用自定义消息头动态发送活动的功能。在我们重试工作进程服务的示例中，我们可以动态更新消息头以用于重试过程。

请注意，MSMQ 4.0 不支持在机器之间移动消息到 DeadLetterQueue。如果在集群中拥有此功能，这将是我们进程执行的一个非常简单的扩展。目前，我们必须考虑这两个队列必须与发送者在同一台机器上。但等等，我们有一个 WCF4 神奇服务来解决这个问题，我强烈建议您在企业复合应用程序中使用它。我最近的文章详细介绍了此服务，例如路由服务此处。

下图显示了一个复合应用程序的示例，该应用程序被分解为小型面向业务的工作流服务，并使用路由服务和 MSMQ 技术进行组合。

基于 FilterTable 的内容，消息可以在同一台机器内或通过 http 通道路由到平衡集群。工作流服务将通过 http 通道将消息发送到具有集群地址的路由器服务。路由器服务（可能是另一个节点，基于负载均衡器设置）会将消息转发到本地队列等。如您所见，通过存储在存储库中的元数据（工作流服务、FilterTable 等），我们可以拥有一个运行时可管理复合应用程序模型。

还有一件事，当我们使用 DeadLetterQueue 时，例如从该队列检索消息，服务地址（监听器 URI）是为我们要从中接收消息的物理队列声明的。如果消息是从其使用者（客户端）发送的，那么一切都将正常。但是，在 DLQ 的情况下，情况有所不同，因为消息是从另一个队列（由另一个客户端、契约等生成）移动过来的。要让 DLQ 服务接收此消息，我们需要调解终结点的 AddressFilter。我们可以为 [ServiceBehavior(AddressFilterMode=AddressFilterMode.Any)] 声明服务行为，或创建自定义 endpointBehavior 来进行特定的 AddressFilter，或者我们可以使用路由服务来实现此神奇的路由，就像在下面的 Test 2 中所做的那样。

以上是关于使用方法的介绍，让我们看一些测试用例来了解动态发送活动的工作原理。

测试

动态发送活动已构建在以下解决方案中

如您所见，有几个项目旨在提供测试支持，例如重新托管的工作流设计器、自托管控制台程序和通用的自托管控制台程序。我将跳过对每个项目的描述，因为您可以在我之前的文章此处中找到有关此测试器的更多信息。

好的，让我们开始 123 个测试。请创建非事务性队列 Test，事务性队列 Search 和 Wait，并为所有队列设置完全权限。

测试 1 - 简单测试

第 1 步。

启动 WorkflowConsoleServiceTester。该程序托管一个从 Test 队列接收消息的服务。接收到的消息显示在屏幕上。

第 2 步。

启动 WorkflowFormsDesignerTester 程序。该工作流设计器将自动加载最后一个工作流。您应该看到以下屏幕

下图显示了属性的值，第一个是消息头的，第二个是 Send 活动的

好的，现在我们准备创建一个测试，例如发送一条带消息头的消息到 Test 队列，并通过控制台程序接收消息。

步骤 3。

按 Run 按钮在自己的托管控制台程序中托管和运行我们的测试工作流。如果一切顺利，您应该在控制台程序中看到以下响应。第一个显示从队列接收到的消息，第二个显示工作流处理。

一旦我们知道测试通过了，我们就可以回到工作流设计器并修改其工作流以进行更复杂的测试，例如使用 DeadLetterQueue、添加 BindingScope 活动等。

测试 2 - 最终测试

此测试将演示所有自定义动态发送活动，通过路由器向服务测试器发送事务性和非事务性的排队消息。测试前，我们需要更改工作流设计器中的 xaml 资源。正如您在解决方案包中所见，有一个 Examples 文件夹，其中包含两个文件。

步骤 1

将 Test2.xaml 文件的内容复制到剪贴板。

第二步

将剪贴板粘贴到工作流设计器的 TreeView 中。为此，请向上移动水平分割条并删除 Activity 节点。下图显示了这些控件

一旦我们的 Test2 内容在 TreeView 中，请按 Load 按钮将 xaml 加载到设计器工作区。

步骤 3。

下图显示了设计器区域中的 Test2。

如您所见，上述测试有三个主要的范围。第一个将排队消息发送到 Wait 队列，并在 TimeToLive 时间过后，消息将移至 Search 队列。第二个范围演示了非事务性排队消息的绑定范围。Endpoint 的地址设置在 Send 活动中。最后一个范围演示了发送事务性消息，其中绑定在 Binding 的 ETB 属性中使用 NetMsmqBinding 类声明。

在此测试中，我还演示了路由功能，作为 MSMQ 和服务测试器之间的集成服务。此连接在以下配置文件中声明性地完成

我建议在您的基础架构中使用路由服务，因为它是非常强大的架构组件。在我们上面的测试中，我们有两个入站端点和一个出站到服务测试器的端点。基于 filterTable，入站消息可以路由到出站端点。在此示例中，所有入站消息都将转发到服务测试器。

步骤 4

按 Run 按钮执行工作流。您应该在服务测试器接收到的控制台程序中看到另外三个消息。

以上是关于动态发送活动的使用和测试。让我们来看看它的实现。

实现

动态发送活动的实现基于从 NativeActivity 派生自定义范围活动，以便能够完全访问工作流运行时模型。为 Send 活动创建范围活动将使我们能够在设计时传递参数，并在运行时使用它们以编程方式处理 Send 活动。

让我们看看 BindingScope 是如何实现的。

首先，以下代码片段显示了一个用于模板化范围和 Send 活动的工厂类

public class BindingScopeFactory : IActivityTemplateFactory
{
  public Activity Create(DependencyObject target)
  {
    return new BindingScope()
     {
         DisplayName = "BindingScope",
         Body = new Send()
         {
             Action = "*",
             OperationName = "ProcessMessage",
             ServiceContractName = "IGenericContract",
             Endpoint = new Endpoint()
             { 
                 Binding = new BasicHttpBinding()                            
             }
         }
     };
  }
}

上述模板将 Send 活动设置为无类型消息和 basicHttpBinding，因此我们没有任何验证问题。注意，这是默认设置，它可以在运行时被 BindingScope 覆盖，如下面的代码片段所示

protected override void Execute(NativeActivityContext context)
{
  // inject headers
  if (this.Headers != null)
  {
    context.Properties.Add("MessageInspector", new SendMessageInspector() 
      { MessageHeaders = this.Headers.Get(context) });
  }

  // update binding
  if (this.Binding != null)
  {
    dynamic dynBinding = this.Binding.Get(context);

    if (dynBinding is XElement || dynBinding is string)
    {
      XElement bindingElement = dynBinding is XElement ? 
        dynBinding : XElement.Parse(dynBinding);
          
      var bs = LibHelper.DeserializeSection<BindingsSection>(
        string.Concat("<bindings>", bindingElement.ToString(), "</bindings>"));
          
      var bce = bs.BindingCollections.Find(b=>b.BindingName==bindingElement.Name.LocalName);

      Binding binding = Activator.CreateInstance(bce.BindingType) as Binding;
        
      if (bce.ConfiguredBindings.Count == 1)
          bce.ConfiguredBindings[0].ApplyConfiguration(binding);

      this.Body.Endpoint.Binding = binding;
    }
    else
    {
      this.Body.Endpoint.Binding = dynBinding;
    }
  }

  context.ScheduleActivity(Body, OnFaulted);
}

如您所见，绑定是动态类型，例如 XElement、String 或 Binding 对象。在 XElement 或 String 的情况下，我们必须将 xml 格式的文本反序列化为 BindingSection 类型，然后我们创建 Binding 对象并应用其配置。一旦我们有了绑定对象，就可以将其分配给 Endpoint。一旦有了自定义 LibHelper.DeserializeSection<T> 函数，这个逻辑就非常直接了。

WF4 具有内置的连接支持，用于在发送消息之前和接收消息之后访问 OperationContext。该概念基于 WCF MessageInspector 的回调。在我们的实现中，实现了 ISendMessageCallback。以下代码片段显示了这个回调，用于将消息头添加到 OutgoingMessageHeaders 集合中

void ISendMessageCallback.OnSendMessage(OperationContext context)
{
  if (this.MessageHeaders is List<MessageHeader>)
  {
    foreach (MessageHeader header in this.MessageHeaders)
    {
      context.OutgoingMessageHeaders.Add(header);
    }
  }
  else if (this.MessageHeaders is MessageHeader)
  {
    context.OutgoingMessageHeaders.Add(this.MessageHeaders);
  }
  else if (this.MessageHeaders is XElement)
  {
    if (string.Compare(this.MessageHeaders.Name.LocalName, "Header", true)==0)
    {
       foreach (var eh in this.MessageHeaders.Elements())
       {
         context.OutgoingMessageHeaders.Add(this.CreateHeader(eh));
       }
    }
    else
    {
      context.OutgoingMessageHeaders.Add(this.CreateHeader(MessageHeaders));
    }
  }
  else
    throw new Exception("Wrong MessageHeader type");
}

请注意，回调对象必须在 NativeActivityContext.Properties 集合中注册（添加）才能调用它。

结论

总之，本文描述了围绕 Send 活动的自定义范围设计模式，但它也可以用于任何序列，例如状态机、嵌套工作流、远程工作流等。使用抽象类 NativeActivity 来创建自定义活动允许我们以编程方式更新其子属性。此外，我还向您展示了如何轻松使用路由服务来集成流程，而不是构建自定义服务行为来进行终结点地址过滤。

WF4 中的动态发送活动

目录

特点

引言

概念与设计

SendScope

BindingScope

SendQueueMessage

用法与测试

测试

测试 1 - 简单测试

第 1 步。

第 2 步。

步骤 3。

测试 2 - 最终测试

步骤 1

第二步

步骤 3。

步骤 4

实现

结论