TPL 完成与完成

Question

我需要从旧数据库导入客户相关数据，并在此过程中执行多次转换。这意味着单个条目需要执行额外的“事件”（同步产品、创建发票等）。

我最初的解决方案是一种简单的并行方法。它工作正常，但有时会出现问题。如果当前处理的客户需要等待相同类型的事件，他们的处理队列可能会被卡住并最终超时，导致每个底层事件也失败（它们依赖于失败的事件）。这种情况并不总是发生，但还是很烦人。

于是我有了另一个想法，分批工作。我的意思是不仅限制同时处理的客户数量，还限制广播到队列的事件数量。在四处寻找想法时，我找到了这个答案，它指向TPL DataFlow。

我做了一个骨架来熟悉它。Complete()我设置了一个简单的管道，但我对和 waiting的用法有点困惑Completion()。

步骤如下

1. 制作一个数字列表（要导入的客户的 ID） - 这是导入逻辑之外的，它只是为了能够触发其余的逻辑
2. 创建一个BatchBlock（能够限制同时处理的客户数量）
3. MyClass1根据 id ( TransformBlock<int, MyClass1>)创建单个项目
4. MyClass2执行一些逻辑并生成( )的集合TransformManyBlock<MyClass1, MyClass2>- 例如，睡眠 1 秒
5. 对集合中的每个项目执行一些逻辑 ( ActionBlock<MyClass2>) - 例如，休眠 1 秒

这是完整的代码：

public static class Program
{
    private static void Main(string[] args)
    {
        var batchBlock = new BatchBlock<int>(2);
        for (var i = 1; i < 10; i++)
        {
            batchBlock.Post(i);
        }


        batchBlock.Complete();
        while (batchBlock.TryReceive(null, out var ids))
        {
            var transformBlock = new TransformBlock<int, MyClass1>(delegate (int id)
            {
                Console.WriteLine($"TransformBlock(id: {id})");
                return new MyClass1(id, "Star Wars");
            });
            var transformManyBlock = new TransformManyBlock<MyClass1, MyClass2>(delegate (MyClass1 myClass1)
            {
                Console.WriteLine($"TransformManyBlock(myClass1: {myClass1.Id}|{myClass1.Value})");
                Thread.Sleep(1000);
                return GetMyClass22Values(myClass1);
            });

            var actionBlock = new ActionBlock<MyClass2>(delegate (MyClass2 myClass2)
            {
                Console.WriteLine($"ActionBlock(myClass2: {myClass2.Id}|{myClass2.Value})");
                Thread.Sleep(1000);
            });
            transformBlock.LinkTo(transformManyBlock);
            transformManyBlock.LinkTo(actionBlock);
            foreach (var id in ids)
            {
                transformBlock.Post(id);
            }

            // this is the point when I'm not 100% sure

            //transformBlock.Complete();
            //transformManyBlock.Complete();
            //transformManyBlock.Completion.Wait();
            actionBlock.Complete();
            actionBlock.Completion.Wait();
        }

        Console.WriteLine();
        Console.WriteLine("Press any key to continue...");
        Console.ReadKey();
    }

    private static IEnumerable<MyClass2> GetMyClass22Values(MyClass1 myClass1)
    {
        return new List<MyClass2>
               {
                   new MyClass2(1, myClass1.Id+ " did this"),
                   new MyClass2(2, myClass1.Id+ " did that"),
                   new MyClass2(3, myClass1.Id+ " did this again")
               };
    }
}

public class MyClass1
{
    public MyClass1(int id, string value)
    {
        Id = id;
        Value = value;
    }

    public int Id { get; set; }

    public string Value { get; set; }
}

public class MyClass2
{
    public MyClass1(int id, string value)
    {
        Id = id;
        Value = value;
    }

    public int Id { get; set; }

    public string Value { get; set; }
}

所以我纠结的一点是结束，我需要打电话Complete()或等待Completion。我似乎找不到合适的组合。我希望看到如下输出：

TransformBlock(id: 1)
TransformBlock(id: 2)
TransformManyBlock(myClass1: 1|Star Wars)
TransformManyBlock(myClass1: 2|Star Wars)
ActionBlock(myClass2: 1|1 did this)
ActionBlock(myClass2: 2|1 did that)
ActionBlock(myClass2: 3|1 did this again)
ActionBlock(myClass2: 1|2 did this)
ActionBlock(myClass2: 2|2 did that)
ActionBlock(myClass2: 3|2 did this again)
TransformBlock(id: 3)
TransformBlock(id: 4)
TransformManyBlock(myClass1: 3|Star Wars)
TransformManyBlock(myClass1: 4|Star Wars)
ActionBlock(myClass2: 1|3 did this)
ActionBlock(myClass2: 2|3 did that)
ActionBlock(myClass2: 3|3 did this again)
ActionBlock(myClass2: 1|4 did this)
ActionBlock(myClass2: 2|4 did that)
ActionBlock(myClass2: 3|4 did this again)

[the rest of the items]


Press any key to exit...   

任何人都可以指出我正确的方向吗？

Answer 1

快到了，您需要调用Complete管道中的第一个块，然后await Completion调用最后一个块。然后在您的链接中，您需要像这样传播完成：

private async static void Main(string[] args) {
    var transformBlock = new TransformBlock<int, MyClass1>(delegate (int id)
    {
        Console.WriteLine($"TransformBlock(id: {id})");
        return new MyClass1(id, "Star Wars");
    });
    var transformManyBlock = new TransformManyBlock<MyClass1, MyClass2>(delegate (MyClass1 myClass1)
    {
        Console.WriteLine($"TransformManyBlock(myClass1: {myClass1.Id}|{myClass1.Value})");
        Thread.Sleep(1000);
        return GetMyClass22Values(myClass1);
    });

    var actionBlock = new ActionBlock<MyClass2>(delegate (MyClass2 myClass2)
    {
        Console.WriteLine($"ActionBlock(myClass2: {myClass2.Id}|{myClass2.Value})");
        Thread.Sleep(1000);
    });

    //propagate completion
    transformBlock.LinkTo(transformManyBlock, new DataflowLinkOptions() { PropagateCompletion = true });
    transformManyBlock.LinkTo(actionBlock, new DataflowLinkOptions() { PropagateCompletion = true});
    foreach(var id in ids) {
        transformBlock.Post(id);
    }


    //Complete the first block
    transformBlock.Complete();

    //wait for completion to flow to the last block
    await actionBlock.Completion;
} 

您还可以将批处理块合并到管道中并消除对调用的需要TryRecieve，但这似乎是流程的另一部分。

编辑

将完成传播到多个块的示例：

public async static void Main(string[] args) {

    var sourceBlock = new BufferBlock<int>();

    var processBlock1 = new ActionBlock<int>(i => Console.WriteLine($"Block1 {i}"));

    var processBlock2 = new ActionBlock<int>(i => Console.WriteLine($"Block2 {i}"));

    sourceBlock.LinkTo(processBlock1);
    sourceBlock.LinkTo(processBlock2);

    var sourceBlockCompletion = sourceBlock.Completion.ContinueWith(tsk => {
        if(!tsk.IsFaulted) {
            processBlock1.Complete();
            processBlock2.Complete();
        } else {
            ((IDataflowBlock)processBlock1).Fault(tsk.Exception);
            ((IDataflowBlock)processBlock2).Fault(tsk.Exception);
        }
    });

    //Send some data...

    sourceBlock.Complete();
    await Task.WhenAll(sourceBlockCompletion, processBlock1.Completion, processBlock2.Completion);
}