代码之家 › 专栏 › 技术社区 › user1912383

TPL数据流执行输出顺序

tpl-dataflow task-parallel-library c#

0

user1912383 · 技术社区 · 8 年前

在下面的测试代码中,我期望得到以下结果:

1, 2000
2, 4000
3, 6000

然而,实际结果是:

3, 6000    
2, 4000
1, 2000

此外,我只在6秒后在屏幕上看到结果。这意味着任何竞争的输入都将等待并处理到下一阶段。

如何使管道在完成输入a后立即输出结果?

    public static void Run()
    {
        Func<int, string> fn = n =>
        {
            var sleep = n * 2000;
            Thread.Sleep(sleep);
            return n + ", " + sleep;
        };

        var opts = new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 4
        };

        var transformBlock = new TransformBlock<int, string>(fn, opts);
        var bufferBlock = new BufferBlock<string>(opts);

        transformBlock.LinkTo(bufferBlock, new DataflowLinkOptions { PropagateCompletion = true });

        for (var i = 3; i > 0; i--)
            transformBlock.Post(i);

        Console.WriteLine(bufferBlock.Receive());
        Console.WriteLine(bufferBlock.Receive());
        Console.WriteLine(bufferBlock.Receive());
    }

1 回复 | 直到 8 年前

1

5

Haukinger 8 年前

默认情况下,即使并行处理,数据流也会保留消息的顺序。

要尽可能快地转换消息(即无序),请设置 EnsureOrdered 到 false 在你的选择中 TransformBlock .

确保使用最新版本的dataflow(目前是nuget软件包) System.Threading.Tasks.Dataflow 存在于版本4.9中),如 重新排序 并不总是存在。

例子:

class Program
{
    static void Main( string[] args )
    {
        var transformBlock = new TransformBlock<int, int>( x =>
        {
            Thread.Sleep( x );
            return x;
        }, new ExecutionDataflowBlockOptions {EnsureOrdered = false, MaxDegreeOfParallelism = 10} );
        var actionBlock = new ActionBlock<int>( x => Console.WriteLine( x ) );
        transformBlock.LinkTo( actionBlock, new DataflowLinkOptions {PropagateCompletion = true});
        for( var i = 9; i > 0; i-- )
            transformBlock.Post( i * 1000 );
        transformBlock.Complete();
        actionBlock.Completion.Wait();
        Console.ReadLine();
    }
}

这个输出