TPL 数据流 - 与事件流一起使用

Question

在我的应用程序中，我几乎每次都有大量来自第三方的事件，我必须处理它们并发送（将它们通过网络发布到组织的 RabbitMQ。我的疑问是关于 TPL 数据流中此处的用法。想象一下代码如下：

 private TransformBlock<QuoteEvent, Quote> _quotesProcessingBlock;

    private ActionBlock<Quote> _deliveryBlock;
    public TplDataFlow()
    {
        _quotesProcessingBlock = new TransformBlock<QuoteEvent, Quote>(
            x => ProcessQuoteEvent(x));
        _deliveryBlock = new ActionBlock<Quote>(quote => Publish(quote));

        _quotesProcessingBlock.LinkTo(
            _deliveryBlock,
             new DataflowLinkOptions { PropagateCompletion = true }
            ); 
    }

    //This callback method registered at the 3rd party events producer.
//It runs single threaded, so I need to process it quickly  
    private void ProcessEvent(QuoteEvent quoteEvent)
    {
        _quotesProcessingBlock.Post(quoteEvent);

        //What will be the trigger for those lines?? 
        _quotesProcessingBlock.Complete();
        _deliveryBlock.Completion.ConfigureAwait(false).GetAwaiter().GetResult();
    }

我不确定何时会使用 _quotesProcessingBlock.Complete(); 和 _deliveryBlock.Completion。在我看来，对每个报价事件（每秒数百次）都这样做是不合理的，

If so should I remove it or should I move it to the other level?

或者，

Dataflow isn't the right solution here?

If so there is a alternative solution?

Answer 1

当您完成该数据流块时，您将调用 Complete。在这种情况下，当没有更多事件要处理时。当您的应用程序关闭时，您可能只执行一次此操作，或者可能根本不执行此操作。

你应该 await 在某些时候 Completion 属性，即使你从未调用 Complete；如果数据流网格失败，它将通知您的代码。

TPL 数据流 - 与事件流一起使用

TPL Data flow - usage with events stream

c#

dataflow

producer-consumer

tpl-dataflow