TPL Dataflow Broadcastblock丢弃最后一条消息

Question

我有一个非常简单的问题。我需要一种方法来轻松地对需要一些时间的消息执行一些处理。在处理过程中，可能会输入新的请求，但除最后一个之外的所有请求都可以被丢弃。

所以我认为TPL Broadcastblock应该这样做，例如查看文档和帖子，以及StackExchange。我创建了以下解决方案并为其添加了一些单元测试，但在单元测试中，有时最后一项不会发送。

这不是我所期望的。如果它应该放弃任何东西，我会说它应该放弃第一项，因为如果它不能处理消息它应该覆盖它的缓冲区1。任何人都可以看到它是什么？
任何帮助将不胜感激！

下面是块代码：

/// <summary> 
/// This block will take items and perform the specified action on it. Any incoming messages while the action is being performed 
/// will be discarded. 
/// </summary> 
public class DiscardWhileBusyActionBlock<T> : ITargetBlock<T> 
{ 
    private readonly BroadcastBlock<T> broadcastBlock; 

    private readonly ActionBlock<T> actionBlock; 

    /// <summary> 
    /// Initializes a new instance of the <see cref="DiscardWhileBusyActionBlock{T}"/> class. 
    /// Constructs a SyncFilterTarget{TInput}. 
    /// </summary> 
    /// <param name="actionToPerform">Thing to do.</param> 
    public DiscardWhileBusyActionBlock(Action<T> actionToPerform) 
    { 
     if (actionToPerform == null) 
     { 
      throw new ArgumentNullException(nameof(actionToPerform)); 
     } 

     this.broadcastBlock = new BroadcastBlock<T>(item => item); 
     this.actionBlock = new ActionBlock<T>(actionToPerform, new ExecutionDataflowBlockOptions { BoundedCapacity = 1, MaxDegreeOfParallelism = 1 }); 
     this.broadcastBlock.LinkTo(this.actionBlock); 
     this.broadcastBlock.Completion.ContinueWith(task => this.actionBlock.Complete()); 
    } 

    public DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept) 
    { 
     return ((ITargetBlock<T>)this.broadcastBlock).OfferMessage(messageHeader, messageValue, source, consumeToAccept); 
    } 

    public void Complete() 
    { 
     this.broadcastBlock.Complete(); 
    } 

    public void Fault(Exception exception) 
    { 
     ((ITargetBlock<T>)this.broadcastBlock).Fault(exception); 
    } 

    public Task Completion => this.actionBlock.Completion; 
} 

下面是用于测试的代码：

[TestClass] 
public class DiscardWhileBusyActionBlockTest 
{ 
    [TestMethod] 
    public void PostToConnectedBuffer_ActionNotBusy_MessageConsumed() 
    { 
     var actionPerformer = new ActionPerformer(); 

     var block = new DiscardWhileBusyActionBlock<int>(actionPerformer.Perform); 
     var buffer = DiscardWhileBusyActionBlockTest.SetupBuffer(block); 

     buffer.Post(1); 

     DiscardWhileBusyActionBlockTest.WaitForCompletion(buffer, block); 

     var expectedMessages = new[] { 1 }; 
     actionPerformer.LastReceivedMessage.Should().BeEquivalentTo(expectedMessages); 
    } 

    [TestMethod] 
    public void PostToConnectedBuffer_ActionBusy_MessagesConsumedWhenActionBecomesAvailable() 
    { 
     var actionPerformer = new ActionPerformer(); 

     var block = new DiscardWhileBusyActionBlock<int>(actionPerformer.Perform); 
     var buffer = DiscardWhileBusyActionBlockTest.SetupBuffer(block); 

     actionPerformer.SetBusy(); 

     // 1st message will set the actionperformer to busy, 2nd message should be sent when 
     // it becomes available. 
     buffer.Post(1); 
     buffer.Post(2); 

     actionPerformer.SetAvailable(); 

     DiscardWhileBusyActionBlockTest.WaitForCompletion(buffer, block); 

     var expectedMessages = new[] { 1, 2 }; 
     actionPerformer.LastReceivedMessage.Should().BeEquivalentTo(expectedMessages); 
    } 

    [TestMethod] 
    public void PostToConnectedBuffer_ActionBusy_DiscardMessagesInBetweenAndProcessOnlyLastMessage() 
    { 
     var actionPerformer = new ActionPerformer(); 

     var block = new DiscardWhileBusyActionBlock<int>(actionPerformer.Perform); 
     var buffer = DiscardWhileBusyActionBlockTest.SetupBuffer(block); 

     actionPerformer.SetBusy(); 

     buffer.Post(1); 
     buffer.Post(2); 
     buffer.Post(3); 
     buffer.Post(4); 
     buffer.Post(5); 

     actionPerformer.SetAvailable(); 

     DiscardWhileBusyActionBlockTest.WaitForCompletion(buffer, block); 

     var expectedMessages = new[] { 1, 5 }; 
     actionPerformer.LastReceivedMessage.Should().BeEquivalentTo(expectedMessages); 
    } 

    private static void WaitForCompletion(IDataflowBlock source, IDataflowBlock target) 
    { 
     source.Complete(); 
     target.Completion.Wait(TimeSpan.FromSeconds(1)); 
    } 

    private static BufferBlock<int> SetupBuffer(ITargetBlock<int> block) 
    { 
     var buffer = new BufferBlock<int>(); 
     buffer.LinkTo(block); 
     buffer.Completion.ContinueWith(task => block.Complete()); 
     return buffer; 
    } 

    private class ActionPerformer 
    { 
     private readonly ManualResetEvent resetEvent = new ManualResetEvent(true); 

     public List<int> LastReceivedMessage { get; } = new List<int>(); 

     public void Perform(int message) 
     { 
      this.resetEvent.WaitOne(TimeSpan.FromSeconds(3)); 
      this.LastReceivedMessage.Add(message); 
     } 

     public void SetBusy() 
     { 
      this.resetEvent.Reset(); 
     } 

     public void SetAvailable() 
     { 
      this.resetEvent.Set(); 
     } 
    } 
} 

Answer 1

当你级别的动作块的BoundedCapacity到1，这意味着，如果它处理完毕，并且已经有它的队列中的项目，它将丢弃消息，这将超出范围。所以基本上发生了什么是你的块做它的工作，在缓冲区已满时拒绝新消息。之后，广播块完成，因为整个消息被发送给收件人，并且它调用Completion，完成整个管道。

您需要检查最后一条消息的返回布尔值Post，或者更可能将最后一条消息存储在某个变量中，以确保它将转至管道。看起来你最好不要使用BroadcastBlock，因为它的目的是将消息的副本提供给链接块的数量，并且只需自己编写你的逻辑。也许你可以用一个简单的BufferBlock来代替。

更新：OfferMessage方法也提供有关正在提供的消息的信息。我认为你根本不需要缓冲块，因为你必须处理管道的非默认逻辑。更容易拥有像_lastMessage这样的字段，存储最后一个值，并在请求被actionBlock接受时将其擦除。您甚至可以完全删除数据流依赖项，因为您只需调用该请求的方法即可。

旁注：你可以link blocks with completion propagation集选项：

var linkOptions = new DataflowLinkOptions { PropagateCompletion = true }; 
this.broadcastBlock.LinkTo(this.actionBlock, linkOptions); 

这可以去除一些你的代码与potentially dangerous ContinueWith使用。如果您需要异步行为，您也可以await broadcastBlock.SendAsync()而不是Post。