为了简单起见,以下我复制了一个LINQPad脚本作为现实生活场景:
var total = 1 * 1000 * 1000;
var cts = new CancellationTokenSource();
var threads = Environment.ProcessorCount;
int capacity = 10;
var edbOptions = new ExecutionDataflowBlockOptions{BoundedCapacity = capacity, CancellationToken = cts.Token, MaxDegreeOfParallelism = threads};
var dbOptions = new DataflowBlockOptions {BoundedCapacity = capacity, CancellationToken = cts.Token};
var gdbOptions = new GroupingDataflowBlockOptions {BoundedCapacity = capacity, CancellationToken = cts.Token};
var dlOptions = new DataflowLinkOptions {PropagateCompletion = true};
var counter1 = 0;
var counter2 = 0;
var delay1 = 10;
var delay2 = 25;
var action1 = new Func<IEnumerable<string>, Task>(async x => {await Task.Delay(delay1); Interlocked.Increment(ref counter1);});
var action2 = new Func<IEnumerable<string>, Task>(async x => {await Task.Delay(delay2); Interlocked.Increment(ref counter2);});
var actionBlock1 = new ActionBlock<IEnumerable<string>>(action1, edbOptions);
var actionBlock2 = new ActionBlock<IEnumerable<string>>(action2, edbOptions);
var batchBlock1 = new BatchBlock<string>(5, gdbOptions);
var batchBlock2 = new BatchBlock<string>(5, gdbOptions);
batchBlock1.LinkTo(actionBlock1, dlOptions);
batchBlock2.LinkTo(actionBlock2, dlOptions);
var bufferBlock1 = new BufferBlock<string>(dbOptions);
var bufferBlock2 = new BufferBlock<string>(dbOptions);
bufferBlock1.LinkTo(batchBlock1, dlOptions);
bufferBlock2.LinkTo(batchBlock2, dlOptions);
var bcBlock = new BroadcastBlock<string>(x => x, dbOptions);
bcBlock.LinkTo(bufferBlock1, dlOptions);
bcBlock.LinkTo(bufferBlock2, dlOptions);
var mainBlock = new TransformBlock<int, string>(x => x.ToString(), edbOptions);
mainBlock.LinkTo(bcBlock, dlOptions);
mainBlock.Dump("Main Block");
bcBlock.Dump("Broadcast Block");
bufferBlock1.Dump("Buffer Block 1");
bufferBlock2.Dump("Buffer Block 2");
actionBlock1.Dump("Action Block 1");
actionBlock2.Dump("Action Block 2");
foreach(var i in Enumerable.Range(1, total))
await mainBlock.SendAsync(i, cts.Token);
mainBlock.Complete();
await Task.WhenAll(actionBlock1.Completion, actionBlock2.Completion);
counter1.Dump("Counter 1");
counter2.Dump("Counter 2");
我对这段代码有两个问题:
- 虽然我将所有适当块的
BoundedCapacity
限制为10个元素,但似乎我几乎可以一次性推送所有1,000,000条消息。这是预期行为吗? - 尽管整个网络都配置为传播完成,但似乎在调用
mainBlock.Complete()
后立即完成了所有块。我希望counter1
和counter2
变量都等于total
。是否有一种方法可以实现这种行为?