我有一个非常简单的问题。我需要一种方法来轻松地对需要一些时间的消息执行一些处理。处理时,可能会输入新的请求,但除了最后一个请求之外的所有请求都可以被丢弃。
所以我认为 TPL Broadcastblock
应该做到这一点,例如查看 StackExchange 上的文档和帖子。我创建了以下解决方案并为其添加了一些单元测试,但在单元测试中,有时没有发送最后一项。
这不是我所期望的。如果它应该删除任何东西,我会说它应该删除第一个项目,因为如果它无法处理消息,它应该覆盖它的缓冲区 1。谁能看出这是什么吗?
任何帮助将不胜感激!
这是该 block 的代码:
/// <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();
}
}
}
最佳答案
当您调平 BoundedCapacity
时操作 block 到 1
,这意味着,如果它进行处理,并且队列中已经有项目,它将丢弃该消息,这将超出范围。所以基本上发生的事情是你的 block 完成它的工作,在缓冲区已满时拒绝新消息。之后,广播 block 完成,整个消息发送给接收者,并调用 Completion
,完成整个管道。
您需要检查 Post
返回的 bool 值以获取最后一条消息,或者更可能将最后一条消息存储在某个变量中,以确保它将进入管道。看起来你最好不要使用 BroadcastBlock
,其目的是提供消息的副本到链接 block 的数量,并且只需自己编写逻辑即可。也许你可以使用一个简单的 BufferBlock
相反。
更新:OfferMessage
方法还提供有关所提供消息的信息。我认为您根本不需要缓冲区 block ,因为您必须处理管道的非默认逻辑。使用像 _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
。
关于c# - TPL 数据流广播 block 丢弃最后一条消息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45408792/