TPL 数据流 - 块未按预期处理
TPL Dataflow - block not processing as expected
我有一组主要以串行方式处理的简单块,但我有两个要并行处理的块(processblock1 和 processblock2)。我刚开始玩 TPL 数据块,对它来说太新了。
但是在下面的代码中,我可以看到 paraellelblock1 被调用但从未像预期的那样调用 parallelblock2。我希望他们能同时开始。
class Program
{
static void Main(string[] args)
{
var readBlock = new TransformBlock<int, int>(x => DoSomething(x, "readBlock"),
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 }); //1
var processBlock1 =
new TransformBlock<int, int>(x => DoSomething(x, "processBlock1")); //2
var processBlock2 =
new TransformBlock<int, int>(x => DoSomething(x, "processBlock2")); //3
var saveBlock =
new ActionBlock<int>(
x => Save(x)); //4
readBlock.LinkTo(processBlock1,
new DataflowLinkOptions { PropagateCompletion = true }); //5
readBlock.LinkTo(processBlock2,
new DataflowLinkOptions { PropagateCompletion = true }); //6
processBlock1.LinkTo(
saveBlock); //7
processBlock2.LinkTo(
saveBlock); //8
readBlock.Post(1); //10
Task.WhenAll(
processBlock1.Completion,
processBlock2.Completion)
.ContinueWith(_ => saveBlock.Complete()); //11
readBlock.Complete(); //12
saveBlock.Completion.Wait(); //13
Console.WriteLine("Processing complete!");
Console.ReadLine();
}
private static int DoSomething(int i, string method)
{
Console.WriteLine($"Do Something, callng method : { method}");
return i;
}
private static async Task<int> DoSomethingAsync(int i, string method)
{
DoSomething(i, method);
return i;
}
private static void Save(int i)
{
Console.WriteLine("Save!");
}
}
默认情况下,tpl 块只会向第一个链接块发送消息。
使用 BroadcastBlock 向多个组件发送消息。
void Main()
{
var random = new Random();
var readBlock = new TransformBlock<int, int>(x => { return DoSomething(x, "readBlock"); },
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 }); //1
var braodcastBlock = new BroadcastBlock<int>(i => i); // ⬅️ Here
var processBlock1 =
new TransformBlock<int, int>(x => DoSomething(x, "processBlock1")); //2
var processBlock2 =
new TransformBlock<int, int>(x => DoSomething(x, "processBlock2")); //3
var saveBlock =
new ActionBlock<int>(
x => Save(x)); //4
readBlock.LinkTo(braodcastBlock, new DataflowLinkOptions { PropagateCompletion = true });
braodcastBlock.LinkTo(processBlock1,
new DataflowLinkOptions { PropagateCompletion = true }); //5
braodcastBlock.LinkTo(processBlock2,
new DataflowLinkOptions { PropagateCompletion = true }); //6
processBlock1.LinkTo(
saveBlock); //7
processBlock2.LinkTo(
saveBlock); //8
readBlock.Post(1); //10
readBlock.Post(2); //10
Task.WhenAll(
processBlock1.Completion,
processBlock2.Completion)
.ContinueWith(_ => saveBlock.Complete());
readBlock.Complete(); //12
saveBlock.Completion.Wait(); //13
Console.WriteLine("Processing complete!");
}
// Define other methods and classes here
private static int DoSomething(int i, string method)
{
Console.WriteLine($"Do Something, callng method : { method} {i}");
return i;
}
private static Task<int> DoSomethingAsync(int i, string method)
{
DoSomething(i, method);
return Task.FromResult(i);
}
private static void Save(int i)
{
Console.WriteLine("Save! " + i);
}
您似乎只向图表发布了一个项目,第一个消费它的消费者获胜。您制作的图表中没有隐含的 'tee' 功能——因此那里不可能存在并行性。
我有一组主要以串行方式处理的简单块,但我有两个要并行处理的块(processblock1 和 processblock2)。我刚开始玩 TPL 数据块,对它来说太新了。 但是在下面的代码中,我可以看到 paraellelblock1 被调用但从未像预期的那样调用 parallelblock2。我希望他们能同时开始。
class Program
{
static void Main(string[] args)
{
var readBlock = new TransformBlock<int, int>(x => DoSomething(x, "readBlock"),
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 }); //1
var processBlock1 =
new TransformBlock<int, int>(x => DoSomething(x, "processBlock1")); //2
var processBlock2 =
new TransformBlock<int, int>(x => DoSomething(x, "processBlock2")); //3
var saveBlock =
new ActionBlock<int>(
x => Save(x)); //4
readBlock.LinkTo(processBlock1,
new DataflowLinkOptions { PropagateCompletion = true }); //5
readBlock.LinkTo(processBlock2,
new DataflowLinkOptions { PropagateCompletion = true }); //6
processBlock1.LinkTo(
saveBlock); //7
processBlock2.LinkTo(
saveBlock); //8
readBlock.Post(1); //10
Task.WhenAll(
processBlock1.Completion,
processBlock2.Completion)
.ContinueWith(_ => saveBlock.Complete()); //11
readBlock.Complete(); //12
saveBlock.Completion.Wait(); //13
Console.WriteLine("Processing complete!");
Console.ReadLine();
}
private static int DoSomething(int i, string method)
{
Console.WriteLine($"Do Something, callng method : { method}");
return i;
}
private static async Task<int> DoSomethingAsync(int i, string method)
{
DoSomething(i, method);
return i;
}
private static void Save(int i)
{
Console.WriteLine("Save!");
}
}
默认情况下,tpl 块只会向第一个链接块发送消息。 使用 BroadcastBlock 向多个组件发送消息。
void Main()
{
var random = new Random();
var readBlock = new TransformBlock<int, int>(x => { return DoSomething(x, "readBlock"); },
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 }); //1
var braodcastBlock = new BroadcastBlock<int>(i => i); // ⬅️ Here
var processBlock1 =
new TransformBlock<int, int>(x => DoSomething(x, "processBlock1")); //2
var processBlock2 =
new TransformBlock<int, int>(x => DoSomething(x, "processBlock2")); //3
var saveBlock =
new ActionBlock<int>(
x => Save(x)); //4
readBlock.LinkTo(braodcastBlock, new DataflowLinkOptions { PropagateCompletion = true });
braodcastBlock.LinkTo(processBlock1,
new DataflowLinkOptions { PropagateCompletion = true }); //5
braodcastBlock.LinkTo(processBlock2,
new DataflowLinkOptions { PropagateCompletion = true }); //6
processBlock1.LinkTo(
saveBlock); //7
processBlock2.LinkTo(
saveBlock); //8
readBlock.Post(1); //10
readBlock.Post(2); //10
Task.WhenAll(
processBlock1.Completion,
processBlock2.Completion)
.ContinueWith(_ => saveBlock.Complete());
readBlock.Complete(); //12
saveBlock.Completion.Wait(); //13
Console.WriteLine("Processing complete!");
}
// Define other methods and classes here
private static int DoSomething(int i, string method)
{
Console.WriteLine($"Do Something, callng method : { method} {i}");
return i;
}
private static Task<int> DoSomethingAsync(int i, string method)
{
DoSomething(i, method);
return Task.FromResult(i);
}
private static void Save(int i)
{
Console.WriteLine("Save! " + i);
}
您似乎只向图表发布了一个项目,第一个消费它的消费者获胜。您制作的图表中没有隐含的 'tee' 功能——因此那里不可能存在并行性。