原文链接：http://blogs.msdn.com/b/masimms/archive/2010/08/10/building-your-first-end-to-end-streaminsight-application.aspx

样例代码下载

• 输入适配器 输入适配器提供了从多种数据源（Web服务，数据库，传感器等）收集数据的接口。StreamInsight输入适配器是实现了一堆接口的.NET类集合。它们可以是特定类型（以下简称特型），也可以是一般类型（以下简称泛型）。特型适配器有一个固定的（即编译期知道的）输入数据类型，而泛型适配器根据配置信息在运行时推断数据类型。例如，SQL Server输入适配器会在运行时根据SELECT语句，存储过程等来推断数据类型。
• 查询 持续查询定义了数据流动过程中的询问问题。查询可以链接在一起，一个完整的端到端的查询是由输入适配器，一系列查询和输出适配器组成的。
• 输出适配器 输出适配器提供了将查询结果传输到外部系统和应用中的接口。它和输入适配器是镜像关系（它也可以是特型或泛型等等）。
•  托管方式 StreamInsight提供了各式各样的部署模型来最好地满足应用或部署。StreamInsight API（也就是所有的StreamInsight代码）既能适用于进程内的“内嵌”StreamInsight服务器，也可以适用于远程服务器（一般来说会托管于Windows服务中，例如随StreamInsight发布的StreamInsightHost.exe）

在这篇博文中，我们会一起打造一个非常简单的端到端的StreamInsight应用，它会包含以下几个组件：

• 模拟数据输入适配器 不断生成带有随机数和随机区间的特型输入事件
• 控制台输出适配器 将输出事件传送到控制台窗口
• 查询 我们会开发多个不同查询来展示查询组合，时间窗口和聚合
• 进程内托管 为了简化，我们使用进程内托管模型

熟悉StreamInsight的读者可能会想到“上面这些适配器都是codeplex上StreamInsight标准样例的一部分”。完全正确！我会带你看到这些适配器是如何构建的（我是作者之一哦:)，然后谈谈一些设计的选择。

编写第一个查询

首先我们先看一个简单的端到端的查询示例，它会用到我们稍后开发的一些组件。

• 打开IntroToStreamInsight项目，从这里可以下载。
• 打开Main.cs文件，代码如下：
  
  

  using System;
  using System.Collections.Generic;
  using System.Linq;
  using System.Text;
  using Microsoft.ComplexEventProcessing.Linq;
  using IntroHost.SimulatedInputAdapter;
  using Microsoft.ComplexEventProcessing;
  using IntroHost.ConsoleOutput;
  
  namespace IntroHost
  {
      class Program
      {
          static void Main(string[] args)
          {
              // 使用app.config文件中的默认设置初始化StreamInsight日志记录器
              StreamInsightLog.Init();
  
              // 使用StreamInsight "Default" 实例
              string instanceName = "Default";
  
              // 嵌入StreamInsight引擎
              using (Server cepServer = Server.Create(instanceName))
              {
                  // 创建一个应用以托管查询和适配器
                  Application cepApplication = cepServer.CreateApplication("simple");
  
                  // 使用SimulatedInput适配器创建输入数据流，
                  // 生成SimpleEventType类型的事件，使用SimpleEventTypeFiller类填充负载字段。
                  // 我们每100毫秒生成一个点事件（每秒10个）
                  var input = CepStream<SimpleEventType>.Create("inputStream",
                      typeof(SimulatedInputFactory), new SimulatedInputAdapterConfig()
                      {
                           CtiFrequency = 1,
                           EventPeriod = 100,
                           EventPeriodRandomOffset = 0,
                           TypeInitializer = typeof(SimpleEventTypeFiller).AssemblyQualifiedName
                      }, 
                      EventShape.Point);
  
                  // 将结果直接送入查询。读取输入值，传入到输出适配器中
                  //var query = from e in input select e;
  
                  // 聚合查询。在一个3秒的窗口内计算同一仪表ID的平均值，最小值，最大值和事件数量
                  var query = from e in input
                                 group e by e.MeterId into meterGroups
                                 from win in meterGroups.HoppingWindow(
                                     TimeSpan.FromSeconds(3),
                                     TimeSpan.FromSeconds(2),
                                     HoppingWindowOutputPolicy.ClipToWindowEnd)
                                 select new
                                 {
                                     meterId = meterGroups.Key,
                                     avg = win.Avg(e => e.Value),
                                     max = win.Max(e => e.Value),
                                     min = win.Min(e => e.Value),
                                     count = win.Count()
                                 };
                                 
  
                  // 将查询绑定到输出适配器，结果输出到控制台
                  var output = query.ToQuery(cepApplication, "simpleQuery", "A Simple Query",
                      typeof(ConsoleAdapterFactory), new ConsoleAdapterConfig()
                      {
                          DisplayCtiEvents = true,
                          SingleLine = true,
                          Target = TraceTarget.Console
                      }, EventShape.Point, StreamEventOrder.FullyOrdered);
  
                  output.Start();
  
                  Console.WriteLine("Query active - press <enter> to shut down");
                  Console.ReadLine();
  
                  output.Stop();
              }
          }
      }
  }
  
  

  最神奇的要数定义query的部分，它读取进入的模拟事件，按照仪表ID进行分组，而后创建一个3秒宽，步长为2秒的前进（创建输出）时间窗口。查询的输出包含了仪表ID、平均值、最小值和最大值。
  执行应用程序会得到类似下面的输出结果（由于这些值都是随机生成的，因此可能会不一样）：
  

  2010-08-09 17:12:09,314 INFO - Advance time policy: CTI f 1, CTI offset -00:00:00.0000001, time policy: Adjust Query active - press to shut down CTI - 12:12:08.000 POINT(12:12:08.000) avg = 57.7530399389036, count = 3, max = 93.6887342453416, meterId = ValveThree, min = 7.61713502351992, POINT(12:12:08.000) avg = 60.2433371172488, count = 2, max = 77.736259381164, meterId = ValveTwo, min = 42.7504148533337, POINT(12:12:08.000) avg = 60.1549154078378, count = 4, max = 84.4917737340982, meterId = ValveOne, min = 13.812115236098, CTI - 12:12:10.000 POINT(12:12:10.000) avg = 46.1892059753599, count = 8, max = 93.6887342453416, meterId = ValveThree, min = 7.61713502351992, POINT(12:12:10.000) avg = 50.7219083455338, count = 6, max = 77.736259381164, meterId = ValveTwo, min = 8.38052840362328, POINT(12:12:10.000) avg = 62.5471921749812, count = 13, max = 94.2385576638573, meterId = ValveOne, min = 3.48205389617107, CTI - 12:12:12.000

  将SingleLine=true改成SingleLine=false，重新运行程序：

  Query active - press to shut down CTI - 12:12:54.000 POINT(12:12:54.000) avg: 29.9929076712545 count: 2 max: 37.6554307703187 meterId: ValveTwo min: 22.3303845721904 POINT(12:12:54.000) avg: 58.6186501121235 count: 4 max: 91.9514771047754 meterId: ValveThree min: 10.0425148895208 POINT(12:12:54.000) avg: 42.7088354447432 count: 4 max: 71.3766915590394 meterId: ValveOne min: 23.3763407093363 CTI - 12:12:56.000

  这么看直观多了！ 我们可以看到结果按照仪表ID进行了分组，接下去对模拟数据进行聚合以得到平均值、最小值和最大值。将第一次运行的结果使用Excel变为表格状如下：
  
  

  Date	Time	Avg	Count	Max	Meter ID	Min
  8/9/2010	21:54:02	13.58	2	16.65	ValveThree	10.52
  8/9/2010	21:54:02	74.14	3	88.43	ValveTwo	64.09
  8/9/2010	21:54:02	45.35	7	88.92	ValveOne	3.18

这看起来还不是很疯狂——你会问那么该怎样构造出这些呢？一般来说，上面的代码是StreamInsight开发中需要花大部分时间做的工作——定义和执行查询。对于某些情形你需要挖掘的更深入一些，如定义你自己的适配器等等。文章的剩下部分将介绍如何从头开始打造底层组件，注意这项工作不是每次都需要做的哦。

深度挖掘——怎样打造组件

下面博文分为以下几个段落：

• 建立 Visual Studio项目
• 构建特型输入适配器生成模拟数据
  • 定义配置
  • 创建适配器工厂
  • 创建适配器实例
  • 格式化事件显示
• 构建泛型输出适配器输出到控制台
  • 定义配置
  • 创建适配器工场
  • 创建适配器实例

建立 Visual Studio项目

1. 在Visual Studio中新建一个C#控制台应用（在下载代码中，我起的名字叫做IntroToStreamInsight）。使用.NET 4.0作为目标框架（记住不是NET Framework 4 Client Profile）。
2. 从GAC中添加下列引用到你的项目中：
   1. Microsoft.ComplexEventProcessing.dll
   2. Microsoft.ComplexEventProcessing.Adapters.dll
   

如果你没有这个时髦的“添加应用（Add Reference）”窗口，请在“扩展器（Extension Manager）”中安装如下插件：

1. 点击Tools，选择Extension Manager；
2. 在扩展管理器窗口中，点击Online Gallery；
3. 按照最高打分排名，选择Productivity Power Tools;
4. 体验一下Add Reference窗口中的搜索以及不同项目间的拷贝和粘贴功能吧。

构建特型输入适配器生成模拟数据

在编写查询之前，我们需要流入一些数据到应用中。这些数据可以来自于多种数据源；为了更快的进行后续工作，一个最简单的方法是使用模拟数据或已有数据。这些数据可以通过录制的数据集（例如.csv文件）重播或者动态生成。在这篇博文中，我们将一起实现一个输入适配器，它可以：

• 实现不同类型的事件（StreamInsight支持三种不同类型事件：点类型事件、间隔类型事件和边缘类型事件，我会在后续的博文中介绍它们之间的不同之处（更多详细内容可以参考这里的事件模型话题），暂时你只要知道以下几点：
  • 点事件 发生在特定的时间点（例如读取传感器）
  • 间隔事件 发生在两个时间点之间（例如两个时间点之间有效的股票报价）
  • 边缘事件 有一个特定的起始时间，但是没有明确的结束时间（例如一个股票报价仍然有效，但是说不定什么时候就不再有效了）
• 生成可在编译期定义的特型事件（如，使用泛型），并可以随机初始化负载数据和事件流动率的间隔（也就是，多久可以生成一个事件）
• 提供可配置的日志功能来帮助诊断和调试（这并不是说我们一定会犯错，而是说这个功能在复杂适配器中会非常方便）

想要了解更多关于开发适配器的背景知识，请查阅MSDN上关于创建输入和输出适配器的主题。

定义配置

创建一个适配器的第一步是定义配置接口（它会被其他所有构成适配器的类所使用）。我们的适配器将包括三个配置属性：

• CTI 频率 当前时间增量（ Current Time Increment，简称CTI)）是用来告诉StreamInsight引擎当前时间的一种方式，它仿佛在说：
  我是适配器，我要告诉你我已经接受到所有截止到时间X的输入数据了。至于在时间X后到达的事件，希望你能够对其负责。
  

CTI 的作用是通知StreamInsight引擎继续处理输入事件。如果引擎中没有CTI的时间增长，它将不会生成任何输出（因为我们没有告诉引擎它已经有了足够的数据可以处理）。CTI 越多，则输出流中的响应越多。一个较低的CTI频率可以是在每个输入数据后发送一个CTI事件，而CTI频率为5可以在每5个事件后发送一个CTI。其实把它称为CTI 周期可能更为合适，但由于时间增长设置中使用的是频率这个词，所以我们也这么用了） ；较少的CTI 将提供更大的吞吐量和较高的延迟（由于引擎可以处理大块数据，因此并不需要过多的CTI）。

其他的配置变量都和适配器中的事件生成速率有关。事件之间的等待时间基本公式定义如下：

事件周期EventPeriod + 随机事件周期偏移量(0 –> Event Period Random Offset)

• 事件周期  适配器生成随机事件的基准周期（比如周期为500毫秒就意味着适配器会每500毫秒生成一个事件）
• 事前周期随机偏移量 事件周期基础之上的最大随机偏移量
• 类型初始化器 为了提供随机字段值，我们要定义一个简单的接口（ITypeInitializer<TPayload>，代码如下），并将程序集限定类型名传入配置（传递真实类型会导致DataContractSerialization出错）

public interface ITypeInitializer<TPayload>
{
    void FillValues(TPayload obj);
}

创建一个名为SimulatedInputAdapter的解决方案目录。在目录中新建一个名为SimulatedInputAdapterConfig的类，代码如下：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace IntroHost.SimulatedInputAdapter
{
    /// <summary>
    /// 这是 SimulatedDataInputFactory的配置类型。
    /// 使用SimulatedInputAdapterConfig类实例来配置输入适配器多久生成事件
    /// </summary>
    public struct SimulatedInputAdapterConfig
    {
        /// <summary>
        /// 多久发送CTI事件(1 = 每个数据事件后发送一个CTI事件)
        /// </summary>
        public uint CtiFrequency { get; set; }

        /// <summary>
        /// 适配器生成模拟事件的基准周期，单位毫秒。
        /// （如果值为500，就意味着适配器会在1秒内生成两次事件，间隔500毫秒）
        /// </summary>
        public int EventPeriod { get; set; }

        /// <summary>
        /// 事件周期之上的最大随机偏移
        /// </summary>
        public int EventPeriodRandomOffset { get; set; }

        /// <summary>
        /// TODO
        /// </summary>
        public string TypeInitializer { get; set; } 
    }
}

创建适配器工厂

应用程序中的适配器并不是直接实例化的，它通常是在查询启动时根据绑定的配置信息由StreamInsight引擎创建。在开发适配器时有几点需要注意：

• 适配器实例可以通过工厂接口创建
• 适配器必须能够基于一个序列化的配置文件初始化
  • 这就是为什么你不能直接传递一个实例引用或代理到一个适配器中。相反，你需要提供基于适配器配置入口的某种查找机制（一般通过单件模式或静态方法实现）

适配器工厂必须为公用类，且必须实现下面的适配器工厂接口之一（具体有适配器类型决定）：

工厂还可以有选择的通过实现ITypedDeclareAdvanceTimeProperties来声明CTI行为。如果你的适配器不需要细粒度的去控制时间增长，那么我们推荐大家实现该接口。

未来我会撰写一些博文来深入探讨创建各种不同适配器的细节；暂时我们先简单创建一个基础类型的输入适配器来生成随机事件（模拟数据）。适配器工厂负责处理跨适配器的工作，根据请求创建对应适配器类型（点类型，间隔类型，边缘类型）的实例。

public class SimulatedInputFactory : ITypedInputAdapterFactory<SimulatedInputAdapterConfig>,
       ITypedDeclareAdvanceTimeProperties<SimulatedInputAdapterConfig>
   {
       public InputAdapterBase Create<TPayload>(SimulatedInputAdapterConfig configInfo, 
           EventShape eventShape)
       {
           throw new NotImplementedException();
       }

       public void Dispose()
       {
           throw new NotImplementedException();
       }

       public AdapterAdvanceTimeSettings DeclareAdvanceTimeProperties<TPayload>(
           SimulatedInputAdapterConfig configInfo, EventShape eventShape)
       {
           throw new NotImplementedException();
       }
   }

using System;
using System.Diagnostics;
using Microsoft.ComplexEventProcessing;
using Microsoft.ComplexEventProcessing.Adapters;

namespace IntroHost.SimulatedInputAdapter
{
    public class SimulatedInputFactory : ITypedInputAdapterFactory<SimulatedInputAdapterConfig>,
        ITypedDeclareAdvanceTimeProperties<SimulatedInputAdapterConfig>  
    {
        internal static readonly string ADAPTER_NAME = "SimulatedInput";
        private static readonly StreamInsightLog trace = new StreamInsightLog(ADAPTER_NAME);
            
        /// <summary>
        /// 基于配置和事件类型生成适配器引用
        /// </summary>
        public InputAdapterBase Create<TPayload>(SimulatedInputAdapterConfig configInfo, 
            EventShape eventShape) 
        {
            InputAdapterBase ret = default(InputAdapterBase);
            switch (eventShape)
            {
                case EventShape.Point:
                    ret = new SimulatedInputPointAdapter<TPayload>(configInfo);
                    break;
            
                case EventShape.Interval:
                    ret = new SimulatedInputIntervalAdapter<TPayload>(configInfo);
                    break;

                case EventShape.Edge:
                    ret = new SimulatedInputEdgeAdapter<TPayload>(configInfo);
                    break;
            }

            return ret;
        }

        /// <summary>
        /// 适配器中没有共享资源——Dispose留空
        /// </summary>
        public void Dispose()
        { }

        /// <summary>
        /// 声明式地增加应用时间（加入CTI）
        /// </summary>
        /// <returns></returns>
        public AdapterAdvanceTimeSettings DeclareAdvanceTimeProperties<TPayload>(
            SimulatedInputAdapterConfig configInfo, EventShape eventShape)
        {
            trace.LogMsg(TraceEventType.Information, 
                "Advance time policy: CTI f {0}, CTI offset {1}, time policy: {2}",
                configInfo.CtiFrequency, TimeSpan.FromTicks(-1), AdvanceTimePolicy.Adjust);

            var timeGenSettings = new AdvanceTimeGenerationSettings(configInfo.CtiFrequency,
                TimeSpan.FromTicks(-1), true);

            return new AdapterAdvanceTimeSettings(timeGenSettings, AdvanceTimePolicy.Adjust);
        }
    }
}

适配器工厂不需要管理共享资源和上下文变换。适配器工厂仅做两件核心工作：

• 基于事件类型生成对应适配器实例
• 定义CTI策略，每出现N个事件就在引擎中增加一次CTI

创建适配器实例

从现在起，让我们添加适配器实例剩下的部分。每一个适配器都会定时的创建事件，填充事件以及将相应类型的事件放入到队列中。填充事件（也就是填充负载字段）是外部插接类的工作。正常情况下我们会使用1~2个lambda表达式用来注入这个逻辑。为了避免设置复杂化，例如有些需要往返序列化的表达式，我们定义了一个叫做ITypeInitializer<TPayload>的接口，并创建了一个叫做SimpleEventTypeFiller的类用来实现该接口：

public interface ITypeInitializer<TPayload>
{
    void FillValues(TPayload obj);

将其与应用中的负载类联合使用：

public class SimpleEventType
{
    public string MeterId { get; set; }
    public double Value { get; set; }
    public DateTime Timestamp { get; set; } 
}

public class SimpleEventTypeFiller : ITypeInitializer<SimpleEventType>
{
    private Random rand = new Random();
         
    private string[] Meters = new string[]
    {
        "ValveOne",
        "ValveTwo",
        "ValveThree"
    };

    public void FillValues(SimpleEventType obj)
    {
        obj.MeterId   = Meters[rand.Next(Meters.Length)];
        obj.Value     = rand.NextDouble() * 100;
        obj.Timestamp = DateTime.Now;
    }
}

这是生成SimpleEventType事件并随机填充负载值的最基本的做法。使用这个接口将模拟输入转变为点类型、间隔类型和边缘类型适配器（下面展示了点类型输入的代码，对于下载代码中的间隔类型和边缘类型，我会在后面解释不同之处）：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.ComplexEventProcessing;
using Microsoft.ComplexEventProcessing.Adapters;
using System.Diagnostics;

namespace IntroHost.SimulatedInputAdapter
{
    /// <summary>
    /// 点类型事件的模拟输入适配器，它会周期性的生成随机数据。
    /// 如果一个类实现了ITypeInitializer<TPayload>接口且在配置中指定，
    /// 那么该类实例可用来填充事件。
    /// </summary>
    /// <typeparam name="TPayload"></typeparam>
    public class SimulatedInputPointAdapter<TPayload> : 
        TypedPointInputAdapter<TPayload> 
    {
        /// <summary>
        /// 保存模拟输入配置
        /// </summary>
        private SimulatedInputAdapterConfig config;

        /// <summary>
        /// 创建一个使用工厂类别名字的日志记录对象
        /// </summary>
        private StreamInsightLog trace = new StreamInsightLog(
            SimulatedInputFactory.ADAPTER_NAME);

        /// <summary>
        /// 用以周期性生成事件的计时器对象
        /// </summary>
        private System.Threading.Timer myTimer;

        /// <summary>
        /// 类型初始化对象
        /// </summary>
        private ITypeInitializer<TPayload> init;

        /// <summary>
        /// 锁对象，用以同步访问生成的事件。
        /// 这主要用在适配器代码调试中（因为计时器会持续不断的被触发，导致了多个线程并行的生成事件）。
        /// </summary>
        private object lockObj = new object();

        /// <summary>
        /// 随机对象，用以创建下一个事件生成时的时间偏移量
        /// </summary>
        private Random rand;

        public SimulatedInputPointAdapter(SimulatedInputAdapterConfig config)
        {
            // 保存配置，并生成配置中指定的类型初始化对象
            this.config = config;
            if (this.config.TypeInitializer != null)
            {
                init = (ITypeInitializer<TPayload>)Activator.CreateInstance(
                    Type.GetType(config.TypeInitializer));
            }

            
        }

        /// <summary>
        /// 所有的事件都是异步的。
        /// 如果计时器触发时适配器处于暂停状态，RaiseEvent 函数会立即退出，因为我们不需要对其负责。
        /// </summary>
        public override void Resume()
        {
        }

        /// <summary>
        /// 创建随机时间间隔生成器以及线程计时器用以调度生成的事件
        /// 计时器在500ms开始以确保所有事情准备好
        /// </summary>
        public override void Start()
        {
            rand = new Random();
            
            myTimer = new System.Threading.Timer(
                new System.Threading.TimerCallback(RaiseEvent),
                null, 500, config.EventPeriod);
        }

        /// <summary>
        /// 当计时器触发时，检查适配器的状态；如果还处于运行状态，生成一个新的模拟事件
        /// </summary>
        /// <param name="state"></param>
        private void RaiseEvent(object state)
        {
            // 确保适配器处于运行状态
            // 如果正处于关闭状态，结束定时器并发送Stopped()信号
            if (AdapterState.Stopping == AdapterState)
            {
                myTimer.Dispose();
                Stopped();
            }
            if (AdapterState.Running != AdapterState)
                return;

            // 分配一个点类型事件用以状态进入的消息数据。
            // 如果无法分配，则退出函数
            lock (lockObj)
            {
                PointEvent<TPayload> currEvent = CreateInsertEvent();
                if (currEvent == null)
                    return;
                currEvent.StartTime = DateTime.Now;

                // 创建一个负载对象，并使用已有的初始化对象填充其字段值
                currEvent.Payload = (TPayload)Activator.CreateInstance(typeof(TPayload));                
                if (init != null) 
                    init.FillValues(currEvent.Payload);

                if (trace.ShouldLog(TraceEventType.Verbose))
                {
                    trace.LogMsg(TraceEventType.Verbose, "INSERT - {0}",
                        currEvent.FormatEventForDisplay(false));
                }

                // 如果事件无法放入队列，则释放其内存并通知适配器可以处理事件了（通过Ready())
                if (EnqueueOperationResult.Full == Enqueue(ref currEvent))
                {                    
                    ReleaseEvent(ref currEvent);
                    Ready();
                }
            }

            // 下一个事件将在now + 时间周期 + 随机偏移 时间后生成
            int nextEventInterval = config.EventPeriod +
                rand.Next(config.EventPeriodRandomOffset);
            myTimer.Change(nextEventInterval, nextEventInterval);
        }
    }
}

这个类的核心概念是从异步数据源（推送）生成事件。许多StreamInsight样例使用单线程来轮询资源，这对于拉取资源的确是一个不错的模式；然而，我们的计时器采用的是推送资源的方式，因此并不需要大量的线程和循环代码。所有的状态管理代码（检查是否停止等等）都是采用异步代理的方式完成（如RaiseEvent）。

currEvent.Payload = (TPayload)Activator.CreateInstance(typeof(TPayload));

  public InputAdapterBase Create<TPayload>(SimulatedInputAdapterConfig configInfo, 
            EventShape eventShape) where TPayload : class, new()

间隔类型事件和边缘类型事件都差不多，只是在定义StartTime和EndTime上略有不同。间隔事件适配器根据随机区间定义EndTime，见下面的代码：

// 分配一个点类型事件用以状态进入的消息数据。
// 如果无法分配，则退出函数
lock (lockObj)
{
    // 下一个事件将在now + 时间周期 + 随机偏移 时间后生成
    int nextEventInterval = config.EventPeriod +
        rand.Next(config.EventPeriodRandomOffset);
    myTimer.Change(nextEventInterval, nextEventInterval);

    IntervalEvent<TPayload> currEvent = CreateInsertEvent();
    if (currEvent == null)
        return;
    currEvent.StartTime = DateTime.Now;
    currEvent.EndTime = currEvent.StartTime.AddMilliseconds(nextEventInterval);

    // 创建一个负载对象，并使用已有的初始化对象填充其字段值
    currEvent.Payload = (TPayload)Activator.CreateInstance(typeof(TPayload));
    if (init != null)
        init.FillValues(currEvent.Payload);

    if (trace.ShouldLog(TraceEventType.Verbose))
    {
        trace.LogMsg(TraceEventType.Verbose, "INSERT - {0}",
            currEvent.FormatEventForDisplay(false));
    }

    // 如果事件无法放入队列，则释放其内存并通知适配器可以处理事件了（通过Ready())
    if (EnqueueOperationResult.Full == Enqueue(ref currEvent))
    {
        ReleaseEvent(ref currEvent);
        Ready();
    }
}

private EdgeEvent<TPayload> lastEvent = null;

/// <summary>
/// 当计时器触发时，检查适配器的状态；如果还处于运行状态，生成一个新的模拟事件
/// </summary>
/// <param name="state"></param>
private void RaiseEvent(object state)
{
    // 确保适配器处于运行状态
    // 如果正处于关闭状态，结束定时器并发送Stopped()信号
    if (AdapterState.Stopping == AdapterState)
    {
        myTimer.Dispose();
        Stopped();
    }
    if (AdapterState.Running != AdapterState)
        return;

    // 分配一个点类型事件用以状态进入的消息数据。
    // 如果无法分配，则退出函数
    lock (lockObj)
    {
        // 创建一个负载对象，并使用已有的初始化对象填充其字段值
        int nextEventInterval = config.EventPeriod +
            rand.Next(config.EventPeriodRandomOffset);
        myTimer.Change(nextEventInterval, nextEventInterval);

        if (lastEvent != null)
        {
            EdgeEvent<TPayload> closeEvent = CreateInsertEvent(EdgeType.End);
            closeEvent.StartTime = lastEvent.StartTime;
            closeEvent.EndTime = DateTimeOffset.Now;
            closeEvent.Payload = lastEvent.Payload;

            // 如果事件无法放入队列，则释放其内存并通知适配器可以处理事件了（通过Ready())
            if (EnqueueOperationResult.Full == Enqueue(ref closeEvent))
            {
                ReleaseEvent(ref closeEvent);
                Ready();
            }

            lastEvent = null;
        }

        EdgeEvent<TPayload> currEvent = CreateInsertEvent(EdgeType.Start);
        if (currEvent == null)
            return;
        currEvent.StartTime = DateTime.Now;
        currEvent.EndTime = currEvent.StartTime.AddMilliseconds(nextEventInterval);

        // 创建一个负载对象，并使用已有的初始化对象填充其字段值
        currEvent.Payload = (TPayload)Activator.CreateInstance(typeof(TPayload));
        if (init != null)
            init.FillValues(currEvent.Payload);

        if (trace.ShouldLog(TraceEventType.Verbose))
        {
            trace.LogMsg(TraceEventType.Verbose, "INSERT - {0}",
                currEvent.FormatEventForDisplay(false));
        }

        // 如果事件无法放入队列，则释放其内存并通知适配器可以处理事件了（通过Ready())
        if (EnqueueOperationResult.Full == Enqueue(ref currEvent))
        {
            ReleaseEvent(ref currEvent);
            Ready();
        }

        // 记住开始的边缘事件
        lastEvent = currEvent;
    }
}

格式化事件显示

嘿，想知道那个神奇的FormatEventForDisplay 函数在哪里吗？打开样例项目，你会看到StreamInsightUtils.cs 文件包含了几个扩展函数：

public static string FormatEventForDisplay<TPayload>(this TypedEvent<TPayload> evt,
    bool verbose)
public static void AddPayloadDetailsList<TPayload>(StringBuilder sb,
    TypedEvent<TPayload> evt)
public static void AddPayloadDetailsRow<TPayload>(StringBuilder sb,
    TypedEvent<TPayload> evt)
public static void AddHeaderRow(StringBuilder sb,
  CepEventType eventType)

构建泛型输出适配器输出到控制台

定义配置

配置中定义了三个关键属性:

• DisplayCtiEvents  是否在输出中包含CTI事件。这可以帮助我们了解StreamInsight引擎何时以及怎样增长时间和处理事件（识别窗口边界——没产生一个CTI都会有一个窗口会被处理）
• SingleLine 按行为中心（单行）还是按照详细列表的方式输出数据
• Target  使用TraceTarget枚举来决定跟踪对象（debug或者控制台）

/// <summary>
/// 可能的跟踪类型
/// </summary>
public enum TraceTarget
{
    /// <summary>
    /// 将消息输出到调试窗口
    /// </summary>
    Debug,

    /// <summary>
    /// 将消息输出到控制台
    /// </summary>
    Console,
}

/// <summary>
/// 跟踪输出适配器的配置结构
/// </summary>
public class ConsoleAdapterConfig
{
    /// <summary>
    /// 指定是否在输出流中显示CTI事件
    /// </summary>
    public bool DisplayCtiEvents { get; set; }

    /// <summary>
    /// 指定使用哪个跟踪输出流
    /// </summary>
    public TraceTarget Target { get; set; }

    /// <summary>
    /// 指定以单行方式还是详细格式输出每个事件
    /// </summary>
    public bool SingleLine { get; set; }
}

注意：这是codeplex上样例的一个“阉割”版，它并没有包含将结果输出到文件或.NET跟踪的功能。

创建适配器工场

public class ConsoleAdapterFactory : IOutputAdapterFactory<ConsoleAdapterConfig>
{
    internal static readonly string APP_NAME = "ConsoleOutput";

    /// <summary>
    /// 创建控制台输出适配器实例，输出结果到.NET调试或控制台窗口
    /// </summary>
    public OutputAdapterBase Create(ConsoleAdapterConfig configInfo, 
        EventShape eventShape, CepEventType cepEventType)
    {
        OutputAdapterBase ret = default(OutputAdapterBase);
        switch(eventShape)
        {
            case EventShape.Point:
                ret = new ConsolePointOutputAdapter(configInfo, cepEventType);
                break;

            case EventShape.Interval:
                ret = new ConsoleIntervalOutputAdapter(configInfo, cepEventType);
                break;

            case EventShape.Edge:
                ret = new ConsoleEdgeOutputAdapter(configInfo, cepEventType);
                break;
        }
        return ret;
    }

    public void Dispose()
    {
        
    }
}

创建适配器实例

想要实现每一个泛型适配器实例，我们需要扩展一下Point/Interval/Edge AdapterBase类，并处理各种Start/Resume方法来从队列中拉取数据并发送到目标数据接收器中（这个例子中，数据接收器是控制台或调试流）。

    internal sealed class ConsoleEdgeOutputAdapter : EdgeOutputAdapter
    {
        private StreamInsightLog trace;
        private ConsoleAdapterConfig config;
        private CepEventType eventType;

        public ConsoleEdgeOutputAdapter(ConsoleAdapterConfig config,
            CepEventType type)
        {
            trace = new StreamInsightLog(ConsoleAdapterFactory.APP_NAME);
            this.config = config;
            this.eventType = type;
        }

        /// <summary>
        /// Start()是在引擎想要让适配器开始产生事件时调用
        /// 它由线程池上的一个线程调用，需尽快释放该线程
        /// </summary>
        public override void Start()
        {
            new Thread(this.ConsumeEvents).Start();
        }

        /// <summary>
        /// Resume()是在Dequeue()调用之后队列情况，且引擎可以继续产生事件时调用
        /// Resume()只会在适配器调用Ready之后被调用
        /// 它由线程池上的一个线程调用，需尽快释放该线程
        /// </summary>
        public override void Resume()
        {
            new Thread(this.ConsumeEvents).Start();
        }

        /// <summary>
        /// 主工作线程函数，负责将事件从队列中取出并将它们发送到输出流
        /// </summary>
        private void ConsumeEvents()
        {
            EdgeEvent currentEvent = default(EdgeEvent);

            try
            {
                while (true)
                {
                    if (AdapterState.Stopping == AdapterState)
                    {
                        Stopped();
                        return;
                    }

                    // 从队列中取出事件。如果取出过程失败，那么适配器状态会变为挂起或停止
                    // 如果是挂起，那么调用Ready()来指定可以继续，否则退出线程
                    if (DequeueOperationResult.Empty == Dequeue(out currentEvent))
                    {
                        Ready();
                        return;
                    }

                    string writeMsg = String.Empty;

                    if (currentEvent.EventKind == EventKind.Insert)
                    {
                        writeMsg = currentEvent.FormatEventForDisplay(eventType,
                            !config.SingleLine);
                    }
                    else if (currentEvent.EventKind == EventKind.Cti)
                    {
                        writeMsg = String.Format("CTI - {0}",
                            currentEvent.StartTime.ToString("hh:mm:ss.fff"));
                    }

                    if (config.Target == TraceTarget.Console)
                        Console.WriteLine(writeMsg);
                    else if (config.Target == TraceTarget.Debug)
                        Debug.WriteLine(writeMsg);

                    // 每一个接收到的事件都应当释放
                    ReleaseEvent(ref currentEvent);
                }
            }
            catch (Exception e)
            {
                trace.LogException(e, "Error in console adapter dequeue");
            }
        }
    }
}

小结

哇，代码有点小多，好吧，深呼吸一下。这不是一个标准的StreamInsight开发过程，因为正常情况开发StreamInsight应用的大部分时间都是连接和配置适配器以及专注于编写查询。这是一个从头开始的端到端的项目。对于大部分应用来说，适配器都可以从codeplex样例中找到。通过此次练习，相信你对那些样例中的代码会有更多的认识。

重点回顾：

• 不需要每次都要做这样的工作来构建非常强大的StreamInsight应用
• 输入和输出适配器可以是特型或泛型的，并且可以根据具体实现运行在推送（异步）模式和拉取（轮询）模式中
• 当前事件增量CTI——它会告诉StreamInsight引擎已经接受到截止时间X的所有数据，并且可以继续往后处理：
  • 如果查询不能输出任何结果，这通常与CTI没有正确加入到数据流中有关
  • CTI可以通过在工厂类中实现ITypedDeclareAdvanceTimeProperties 接口来自动加入到流中
• 适配器准备就绪的话，开发和绑定查询会非常顺利

作者: StreamInsight 发表于 2011-07-26 08:08 原文链接