欢迎光临
我们一直在努力

像Labview一样,使用C#构建测量数据流式处理框架

原文引自segmentfault网——郭晓忠的专栏

1. C# DataFlow介绍

官方解释

TPL(任务并行库) 数据流库向具有高吞吐量和低滞后时间的占用大量 CPU 和 I/O 操作的应用程序的并行化和消息传递提供了基础。 它还能显式控制缓存数据的方式以及在系统中移动的方式。传统编程模型通常需要使用回调和同步对象(例如锁)来协调任务和访问共享数据。在数据流模型下,您可以声明当数据可用时的处理方式,以及数据之间的所有依赖项。 由于运行时管理数据之间的依赖项,因此通常可以避免这种要求来同步访问共享数据。 此外,因为运行时计划基于数据的异步到达,所以数据流可以通过有效管理基础线程提高响应能力和吞吐量。

借助于异步消息传递与管道,它可以提供比线程池更好的控制,也比手工线程方式具备更好的性能。我们常常可以消息传递,生产-消费模式或Actor-Agent模式中使用。在TDF是构建于Task Parallel Library (TPL)之上的,它是我们开发高性能,高并发的应用程序的又一利器。

TDP的主要作用就是Buffering Data和Processing Data,在TDF中,有两个非常重要的接口,ISourceBlock<T> 和ITargetBlock<T>接口。继承于ISourceBlock<T>的对象时作为提供数据的数据源对象-生产者,而继承于ITargetBlock<T>接口类主要是扮演目标对象-消费者。在这个类库中,System.Threading.Tasks.Dataflow名称空间下,提供了很多以Block名字结尾的类,ActionBlock,BufferBlock,TransformBlock,BroadcastBlock等9个Block,我们在开发中通常使用单个或多个Block组合的方式来实现一些功能。

1.1    9个典型Block的使用

1.1.1 BufferBlock

BufferBlock是TDF中最基础的Block。BufferBlock提供了一个有界限或没有界限的Buffer,该Buffer中存储T。该Block很像BlockingCollection<T>。可以用过Post往里面添加数据,也可以通过Receive方法阻塞或异步的的获取数据,数据处理的顺序是FIFO的。它也可以通过Link向其他Block输出数据。

private static BufferBlock<int> m_buffer = new BufferBlock<int>();// Producerprivate static void Producer(){    while(true)    {        int item = Produce();        m_buffer.Post(item);    }}// Consumerprivate static void Consumer(){    while(true)    {        int item = m_buffer.Receive();        Process(item);    }}// Mainpublic static void Main(){    var p = Task.Factory.StartNew(Producer);    var c = Task.Factory.StartNew(Consumer);    Task.WaitAll(p,c);}

1.1.2 ActionBlock

ActionBlock实现ITargetBlock,说明它是消费数据的,也就是对输入的一些数据进行处理。它在构造函数中,允许输入一个委托,来对每一个进来的数据进行一些操作。如果使用Action(T)委托,那说明每一个数据的处理完成需要等待这个委托方法结束,如果使用了Func<TInput, Task>)来构造的话,那么数据的结束将不是委托的返回,而是Task的结束。默认情况下,ActionBlock会FIFO的处理每一个数据,而且一次只能处理一个数据,一个处理完了再处理第二个,但也可以通过配置来并行的执行多个数据。

public ActionBlock<int> abSync = new ActionBlock<int>((i) =>            {                Thread.Sleep(1000);                Console.WriteLine(i + " ThreadId:" + Thread.CurrentThread.ManagedThreadId + " Execute Time:" + DateTime.Now);            }        );        public void TestSync()        {            for (int i = 0; i < 10; i++)            {                abSync.Post(i);            }            Console.WriteLine("Post finished");        }

1.1.3 TransformBlock

TransformBlock是TDF提供的另一种Block,顾名思义它常常在数据流中充当数据转换处理的功能。在TransformBlock内部维护了2个Queue,一个InputQueue,一个OutputQueue。InputQueue存储输入的数据,而通过Transform处理以后的数据则放在OutputQueue,OutputQueue就好像是一个BufferBlock。最终我们可以通过Receive方法来阻塞的一个一个获取OutputQueue中的数据。TransformBlock的Completion.Wait()方法只有在OutputQueue中的数据为0的时候才会返回。

举个例子,我们有一组网址的URL,我们需要对每个URL下载它的HTML数据并存储。那我们通过如下的代码来完成:

public TransformBlock<string, string> tbUrl = new TransformBlock<string, string>((url) =>        {            WebClient webClient = new WebClient();            return webClient.DownloadString(new Uri(url));        }        public void TestDownloadHTML()        {            tbUrl.Post("");            tbUrl.Post("www.sina.com.cn");            string baiduHTML = tbUrl.Receive();            string sinaHTML = tbUrl.Receive();        }

2.1 测试需求

我们需要采集三个通道的数据x,y,z. 然后使用x,y,z组合经过如下公式计算得到一个结果m

然后对m数列做中值滤波,每5个数值求中间值,生成一个最终的值。

得到最终值之后,在界面中显示波形,存二进制文件,通过网络发送到数据服务器。

其流式处理图如下:

2.2 业务分析

按照C# Dataflow的思想,流式处理的各个节点可以使用的Block如下图:

细心的人可以看到,最后的业务处理前面加上了一个BroadCastBlock,是为了同时给三个业务分发消息。

2.3 业务实现

2.3.1 架构设计

采用WPF窗体框架,界面如下

2.3.2 代码实现

由于手上确实没有合适的板卡做测试,我就用三个Task模拟数据生成,然后放入三个ConcurrentQueue.

代码如下:

/// <summary>/// 通道1队列/// </summary>private readonly ConcurrentQueue<double> _queue1 = new ConcurrentQueue<double>();/// <summary>/// 通道2队列/// </summary>private readonly ConcurrentQueue<double> _queue2 = new ConcurrentQueue<double>();/// <summary>/// 通道3队列/// </summary>private readonly ConcurrentQueue<double> _queue3 = new ConcurrentQueue<double>();/// <summary>/// 生成通道1数据按钮事件/// </summary>/// <param name="sender"></param>/// <param name="e"></param>private void GenerateChannel1Button_OnClick(object sender, RoutedEventArgs e){    Task.Factory.StartNew(() => GenerateData(this._queue1));}/// <summary>/// 生成通道2数据按钮事件/// </summary>/// <param name="sender"></param>/// <param name="e"></param>private void GenerateChannel2Button_OnClick(object sender, RoutedEventArgs e){    Task.Factory.StartNew(() => GenerateData(this._queue2));}/// <summary>/// 生成通道3数据按钮事件/// </summary>/// <param name="sender"></param>/// <param name="e"></param>private void GenerateChannel3Button_OnClick(object sender, RoutedEventArgs e){    Task.Factory.StartNew(() => GenerateData(this._queue3));}/// <summary>/// 生成数据/// </summary>/// <param name="queue"></param>/// <returns></returns>private async Task GenerateData(ConcurrentQueue<double> queue){    var random = new Random();    while (this._stop)    {        queue.Enqueue(random.NextDouble() * 10);        await Task.Delay(TimeSpan.FromMilliseconds(50));    }}

初始化各种Block

/// <summary>/// 通道1BufferBlock/// </summary>private BufferBlock<double> _bufferBlock1 = new BufferBlock<double>();/// <summary>/// 通道2BufferBlock/// </summary>private BufferBlock<double> _bufferBlock2 = new BufferBlock<double>();/// <summary>/// 通道3BufferBlock/// </summary>private BufferBlock<double> _bufferBlock3 = new BufferBlock<double>();/// <summary>/// 拼接3个通道JoinBlock/// </summary>private JoinBlock<double, double,double> _joinBlock = new JoinBlock<double, double, double>();/// <summary>/// 计算M的TransformBlock/// </summary>private TransformBlock<Tuple<double,double,double>, double> _calculateMTransformBlock =    new TransformBlock<Tuple<double, double, double>, double>(t => t.Item1 * t.Item2 + t.Item3);/// <summary>/// 每5个m组成一组BatchBlock/// </summary>private BatchBlock<double> _mBatchBlock = new BatchBlock<double>(5);/// <summary>/// m的中值滤波TransformBlock/// </summary>private TransformBlock<double[], double> _mMiddleFilterTransformBlock = new TransformBlock<double[], double>(    t =>    {        Array.Sort(t);        return t[2];    });/// <summary>/// 广播mBroadcastBlock/// </summary>private BroadcastBlock<double> _broadcastBlock = new BroadcastBlock<double>(t => t);/// <summary>/// 界面显示ActionBlock/// </summary>private ActionBlock<double> _showPlotActionBlock;/// <summary>/// 写入文件ActionBlock/// </summary>private ActionBlock<double> _writeFileActionBlock;/// <summary>/// 网络上传ActionBlock/// </summary>private ActionBlock<double> _netUpActionBlock;

由于Lambda 需要访问外部变量,则需要在laod事件中初始化:

//UI显示ActionBlockthis._showPlotActionBlock = new ActionBlock<double>(t =>{    if (this.Datas.Count >= 10000)    {        this.Datas.RemoveAt(0);    }    this.Datas.Add(new DataPoint(_xIndex++, t));    Application.Current.Dispatcher.Invoke(() => { Plot?.InvalidatePlot(); });}, new ExecutionDataflowBlockOptions(){    TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()});//写入文件this._writeFileActionBlock = new ActionBlock<double>(t =>{    this._binaryWriter.Write(t);});//上传数据,暂时不实现this._netUpActionBlock = new ActionBlock<double>(t => Console.WriteLine($@"Net upload value: {t}"));

链接这些Block

o/// <summary>/// 链接Blocks/// </summary>private void LinkBlocks()

{    this._bufferBlock1.LinkTo(this._joinBlock.Target1);    this._bufferBlock2.LinkTo(this._joinBlock.Target2);    this._bufferBlock3.LinkTo(this._joinBlock.Target3);    this._joinBlock.LinkTo(this._calculateMTransformBlock);    this._calculateMTransformBlock.LinkTo(this._mBatchBlock);    this._mBatchBlock.LinkTo(this._mMiddleFilterTransformBlock);    this._mMiddleFilterTransfrmBlock.LinkTo(this._broadcastBlock);    this._broadcastBlock.LinkTo(this._showPlotActionBlock);    this._broadcastBlock.LinkTo(this._writeFileActionBlock);    this._broadcastBlock.LinkTo(this._netUpActionBlock);}

开始测量按钮事件:

this._stop = false;this._fileStream = new FileStream($"{DateTime.Now:yyyy_MM_dd_HH_mm_ss}.dat", FileMode.OpenOrCreate,    FileAccess.Write);this._binaryWriter = new BinaryWriter(this._fileStream);Task.Factory.StartNew(async () =>{    while (!this._stop)    {        if (this._queue1.Count > 0)        {            double result;            this._queue1.TryDequeue(out result);            this._bufferBlock1.Post(result);        }        else        {            await Task.Delay(TimeSpan.FromMilliseconds(30));        }    }});Task.Factory.StartNew(async () =>{    while (!this._stop)    {        if (this._queue2.Count > 0)        {            double result;            this._queue2.TryDequeue(out result);            this._bufferBlock2.Post(result);        }        else        {            await Task.Delay(TimeSpan.FromMilliseconds(30));        }    }});Task.Factory.StartNew(async () =>{    while (!this._stop)    {        if (this._queue3.Count > 0)        {            double result;            this._queue3.TryDequeue(out result);            this._bufferBlock3.Post(result);        }        else        {            await Task.Delay(TimeSpan.FromMilliseconds(30));        }    }});

结束测量事件

this._stop = true;

this._binaryWriter.Flush();

this._binaryWriter.Close();

this._fileStream.Close();

最终的效果:

2.3.3 效果体验:

(1)先点击三个生个数据按钮,然后开始测量,可以看到数据马上就会显示,同时会保存,也会上传数据。停止测量后可以看到数据保存到了文件中。

(2)先点击生成通道1和2生成数据,不点击通道3,然后点击开始测量,可以看到没有反应,再点击通道3,就有数据了,根据我们的测量逻辑,这个是对的。

简仪科技微信每周持续更新,敬请期待。

如有任何问题欢迎与我们联系。

赞(98)
分享到: 更多

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址