我有这个代码:
var data = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 1 });
var action = new ActionBlock<int>(async id =>
{
Console.WriteLine("[{0:T}] #{1}: Start", DateTime.Now, id);
await Task.Delay(1000);
Console.WriteLine("[{0:T}] #{1}: End", DateTime.Now, id);
}, new ExecutionDataflowBlockOptions
{
BoundedCapacity = 1,
MaxDegreeOfParallelism = -1
});
data.LinkTo(action, new DataflowLinkOptions { PropagateCompletion = true });
for (var id = 1; id <= 3; id++)
{
Console.WriteLine("[{0:T}] Sending {1}", DateTime.Now, id);
data.SendAsync(id).Wait();
Console.WriteLine("[{0:T}] Sending {1} complete", DateTime.Now, id);
}
data.Complete();
Task.WhenAll(data.Completion, action.Completion).Wait();
这段代码给我这个输出:
[22:31:22] Sending 1
[22:31:22] Sending 1 complete
[22:31:22] Sending 2
[22:31:22] #1: Start
[22:31:22] Sending 2 complete
[22:31:22] Sending 3
[22:31:23] #1: End
[22:31:23] #2: Start
[22:31:23] Sending 3 complete
[22:31:24] #2: End
[22:31:24] #3: Start
[22:31:25] #3: End
为什么不是ActionBlock
并行工作,即使它有无限的 DOP?
原因是你的ActionBlock
似乎并行度有限是因为它有一个BoundedCapacity
of 1. BoundedCapacity
(不像InputCount
) 包括当前正在处理的项目。这可以很容易地证明:
var block = new ActionBlock<int>(_ => Task.Delay(-1), new ExecutionDataflowBlockOptions
{
BoundedCapacity = 1,
MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
});
await block.SendAsync(4); // Adds a new item
await block.SendAsync(4); // Blocks forever
这意味着当您设置MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
该块当时不能接受多个项目,因此实际上限制了并行度。
您可以通过设置更大的值来解决这个问题BoundedCapacity
:
var action = new ActionBlock<int>(async id =>
{
Console.WriteLine("[{0:T}] #{1}: Start", DateTime.Now, id);
await Task.Delay(1000);
Console.WriteLine("[{0:T}] #{1}: End", DateTime.Now, id);
}, new ExecutionDataflowBlockOptions
{
BoundedCapacity = 10,
MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
});
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)