Preciso importar dados relacionados ao cliente do banco de dados herdado e executar várias transformações durante o processo. Isso significa que uma única entrada precisa executar "eventos" adicionais (sincronizar produtos, criar faturas etc.).

Minha solução inicial foi uma abordagem paralela simples. Funciona bem, masas vezes tem problemas. Se os clientes atualmente processados precisarem esperar pelo mesmo tipo de eventos, suas filas de processamento poderão ficar paralisadas e, eventualmente, atingir o tempo limite, causando a falha de todos os eventos subjacentes (eles dependem do que falhou). Isso não acontece o tempo todo, mas é irritante.

Então, eu tive outra idéia, trabalhar em lotes. Quero dizer, não apenas limitar o número de clientes sendo processados ao mesmo tempo, mas também o número de eventos que são transmitidos para as filas. Enquanto procurava por idéias, descobriesta resposta, que aponta para oTPL DataFlow.

Eu fiz um esqueleto para me familiarizar com ele. Eu montei um pipeline simples, mas estou um pouco confuso sobre o uso deComplete() e aguardandoCompletion().

Os passos são os seguintes

Faça uma lista de números (os IDs dos clientes a serem importados) - isso está fora da lógica de importação, apenas para poder acionar o restante da lógicaCrie umBatchBlock (para poder limitar o número de clientes a serem processados ao mesmo tempo)Crie um únicoMyClass1 item com base no ID (TransformBlock<int, MyClass1>)Execute alguma lógica e gere uma coleção deMyClass2 (TransformManyBlock<MyClass1, MyClass2>) - por exemplo, durma por 1 segundoExecute alguma lógica em cada item da coleção (ActionBlock<MyClass2>) - por exemplo, durma por 1 segundo

Aqui está o código completo:

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; }
}

Então, o ponto com o qual luto é o fim, onde eu precisaria ligarComplete() ou aguardeCompletion. Não consigo encontrar a combinação certa. Eu gostaria de ver uma saída da seguinte maneira:

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...   

Alguém pode me indicar a direção certa?