Rx、TPL 数据流和 Windows 服务 [关闭]答案

【问题标题】：Rx, TPL dataflow and windows service [closed]Rx、TPL 数据流和 Windows 服务 [关闭]
【发布时间】：2018-04-19 12:29:23
【问题描述】：

将 Rx 和数据流与不断工作的 Windows 服务一起使用是个好主意吗？我只是想知道使用 rx 间隔每 5 分钟从 db 获取数据并将其放入阻塞收集，然后使用 Tpl 数据流来处理数据。是正确的方式吗？

更新

我有一个新问题。我必须从数据库中获取项目，对其进行验证并在数据库上执行提交。即使验证返回 false 我也必须执行提交。现在，我想每隔 5 分钟从 db 获取数据，但我不想在 dataFlow 中有双行。所以也许我必须使用blockingCollection 来保留行ID？

【问题讨论】：

这可能是许多方法之一，是的，它应该可以工作。是什么阻止了你做一个小的概念证明？
严重缺乏信息，我们怎么可能知道
无论如何，偏离主题，基于意见
其实没有。当您有 dataflow 块时，为什么要将 anything 放入阻塞集合中？只需将数据发布到块中即可。无论如何，Rx 都用于事件处理，而不是计时器。从数据库读取应该是管道的第一个 TransformManyBlock 的工作，而不是 Rx.Second 块应该从第一个块接收数据并做任何你想做的事情
也就是说，您可以链接可观察对象和数据流块。您可以在一个块上调用.AsObserver() 以获取可以订阅可观察对象的观察者，例如var obs=myBlock.AsObserver(); Observable.Interval(..).Subscribe(obs);

标签： c# windows-services system.reactive tpl-dataflow

【解决方案1】：

不完全是。

首先，您根本不需要 BlockingCollection。数据流块有自己的线程安全输入缓冲区。

其次，块也意味着使用它们自己的任务来执行，这使得它们适合运行诸如调用数据库和处理它们的结果之类的操作。另一方面，Rx 旨在处理通常使用 single 线程完成的事件流。

也就是说，您可以组合 Rx 和 Dataflow 块。我正在使用 Dataflow 管道来

其次，Rx 是用来处理事件流的。为此，它通常在 current 线程上运行。

也就是说，您可以组合 Rx 和数据流块。将区间序列与块结合起来很容易：

var block=new ActionBlock<int>(i=>whatever(i));
var blockObs=blockObs.AsObserver();
var interval=Observable.Interval(obs).Subscribe(obs);

使用数据库需要更多块。假设您要处理单个行，第一个块应该接收触发器、加载数据并返回行。第二个块应该接收各个行并处理它们。

假设您使用 Dapper 将行作为对象返回：

 var headBlock=new TransformManyBlock<int,SomeRow>(_=>
    {
        using(var con=new SqlConnection(..whatever))
        {
            var items=con.ExecuteQuery(theQuery);
            return itmes;
        }
    });

 var secondBlock = new ActionBlock<SomeRow>(row=>DoSomething(row), 
                          new ExecutionDatalfowBlockOptions{MaxDegreeOfParalelism=10});

 headBlock.LinkTo(secondBlock,new DataFlowLinkOptions{PropagateCompletion=true});

 var headObs=headBlock.AsObserver();

 Observable.Interval(TimeSpan.FromMinutes(1)).Subscribe(headObs);

不过，这只是一个演示。当您关闭服务器并等待任何待处理的操作完成时，您应该添加代码以停止管道。

您还应该添加错误处理代码，即使发生错误，您也可以继续处理。如果未处理异常，则抛出的块终止。如果您返回“包装”结果，您可以处理此问题，并带有指示成功或失败的标志。 LinkTo 接受一个谓词，您可以使用该谓词将失败的消息/行移动到记录器或丢弃它们，例如：

var loggerBlock=new ActionBlock<RowWithFlag>(row=>_log.Error(...));

headBlock.LinkTo(secondBlock,linkOptions,row=>row.IsOK);
headBlock.LinkTo(loggerBlock,linkOptions,row=>!row.IsOK);

【讨论】：