Verknüpfen von dynamisch erstellten ActionBlocks mit einem BufferBlock

Question

Ich bin mir nicht sicher, ob das möglich ist, aber wenn es so ist, mache ich es wahrscheinlich nicht richtig. Nehmen wir an, ich habe einen gemeinsamen Puffer, der mit vielen Verbrauchern verbunden ist (ActionBlocks). Jeder Verbraucher sollte Daten konsumieren, die ein Prädikat erfüllen, das verwendet wird, um es mit dem Puffer zu verknüpfen. Zum Beispiel sollte ActionBlock1 Zahlen verbrauchen, die x => x % 5 == 0 erfüllen, sollten ActionBlock2 verbrauchen nur x => x % 5 == 1 usw.

Hier ist, was ich habe:

private static ITargetBlock<int> BuildPipeline(int NumProductionLines) 
{ 
    var productionQueue = new BufferBlock<int>(); 

    for (int i = 0; i < NumProductionLines; i++) 
    { 
     ActionBlock<int> productionLine = new ActionBlock<int>(num => Console.WriteLine("Processed by line {0}: {1}", i + 1, num)); 

     productionQueue.LinkTo(productionLine, x => x % NumProductionLines == i); 
    } 

    return productionQueue; 
} 

Und dann rufe ich:

Random rnd = new Random(); 

ITargetBlock<int> temp = BuildPipeline(5); 

while (true) 
{ 
    temp.Post(rnd.Next(255)); 
} 

Doch diese funktioniert nicht. In der Konsole wird keine Ausgabe angezeigt. Wenn i BuildPipeline Methode ändern, wie:

private static ITargetBlock<int> BuildPipeline(int NumProductionLines) 
{ 
    var productionQueue = new BufferBlock<int>(); 

    ActionBlock<int> productionLine1 = new ActionBlock<int>(num => Console.WriteLine("Processed by line {0}: {1}", 1, num)); 
    ActionBlock<int> productionLine2 = new ActionBlock<int>(num => Console.WriteLine("Processed by line {0}: {1}", 2, num)); 
    ActionBlock<int> productionLine3 = new ActionBlock<int>(num => Console.WriteLine("Processed by line {0}: {1}", 3, num)); 
    ActionBlock<int> productionLine4 = new ActionBlock<int>(num => Console.WriteLine("Processed by line {0}: {1}", 4, num)); 
    ActionBlock<int> productionLine5 = new ActionBlock<int>(num => Console.WriteLine("Processed by line {0}: {1}", 5, num)); 

    productionQueue.LinkTo(productionLine1, x => x % 5 == 0); 
    productionQueue.LinkTo(productionLine2, x => x % 5 == 1); 
    productionQueue.LinkTo(productionLine3, x => x % 5 == 2); 
    productionQueue.LinkTo(productionLine4, x => x % 5 == 3); 
    productionQueue.LinkTo(productionLine5, x => x % 5 == 4); 

    return productionQueue; 
} 

der Code tut, was es ist zu erwarten, zu tun.

Kann jemand herausfinden, warum das dynamische Erstellen und Verknüpfen von Aktionsblöcken nicht funktioniert?

P.S. Wenn ich direkt nach ITargetBlock<int> temp = BuildPipeline(5); Code einbringe, zeigt Temp an, dass 5 Ziele mit dem Puffer verbunden sind. Und die ID jedes Ziels ist anders.

Vielen Dank im Voraus

EDIT: Hinzugefügt Änderungen von svick vorgeschlagen, aber noch nicht gut:

private static ITargetBlock<int> BuildPipeline(int NumProductionLines) 
{ 
    var productionQueue = new BufferBlock<int>(); 
    var opt = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 }; 

    for (int i = 0; i < NumProductionLines; i++) 
    { 
     ActionBlock<int> productionLine = new ActionBlock<int>(num => Console.WriteLine("Processed by line {0}: {1}", i + 1, num)); 

     int j = i; 
     productionQueue.LinkTo(productionLine, x => x % NumProductionLines == j); 
    } 

    ActionBlock<int> discardedLine = new ActionBlock<int>(num => Console.WriteLine("Discarded: {0}", num)); 
    productionQueue.LinkTo(discardedLine); 

    return productionQueue; 
} 

Jetzt zweiten Produktionslinie verarbeitet Daten nur (das gilt: x% 5 == 1 Prädikat). Und die Daten nicht erfüllt das Prädikat, das heißt, ich Zahlen bekommen endet in 9 und 7.

EDIT: Code Arbeiten in etwa wie folgt aussehen:

private static ITargetBlock<int> BuildPipeline(int NumProductionLines) 
{ 
    var productionQueue = new BufferBlock<int>(); 
    var opt = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 }; 

    for (int i = 0; i < NumProductionLines; i++) 
    { 
     int j = i; 
     ActionBlock<int> productionLine = new ActionBlock<int>(num => Console.WriteLine("Processed by line {0}: {1}", j + 1, num)); 

     productionQueue.LinkTo(productionLine, x => x % NumProductionLines == j); 
    } 

    productionQueue.LinkTo(DataflowBlock.NullTarget<int>()); 

    return productionQueue; 
} 

Answer 1

Das Problem ist, dass in der ersten Version, Sie verwenden für jeden Zielblock dasselbe Prädikat. Mit anderen Worten, das Prädikat hängt nicht von i ab.

Aber selbst wenn, würde Ihr Code nicht funktionieren, weil die i Variable unter den Prädikaten geteilt wird, so dass sie alle den letzten Wert verwenden. Das Problem besteht darin, i in eine lokale Variable zu kopieren und diese im Prädikat zu verwenden.

Der Code könnte wie folgt aussehen:

private static ITargetBlock<int> BuildPipeline(int NumProductionLines) 
{ 
    var productionQueue = new BufferBlock<int>(); 

    for (int i = 0; i < NumProductionLines; i++) 
    { 
     int iCopy = i; 

     ActionBlock<int> productionLine = new ActionBlock<int>(
      num => Console.WriteLine("Processed by line {0}: {1}", iCopy + 1, num)); 

     productionQueue.LinkTo(
      productionLine, x => x % NumProductionLines == iCopy); 
    } 

    return productionQueue; 
} 

Wenn Sie sich fragen, warum nicht Ihren Code Prozess zumindest die x % 5 == 1 Zahlen, das ist, weil der Zufallsgenerator wahrscheinlich eine Reihe generieren, die nicht der Fall ist Passen Sie dieses Prädikat an, damit keiner der ActionBlock s es akzeptiert. Aus diesem Grund bleibt die Nummer in der Ausgabewarteschlange des Quellblocks und andere Nummern können nicht durchgehen.

Wenn in Ihrem realen Code, könnten ähnliche Situation wollen passieren und Sie alle Zahlen verwerfen, die eine der Prädikate nicht passen, können Sie den Quellenblock zu a block that does nothing verknüpfen und alles akzeptiert, nachdem Sie es auf alle verlinken Ihre nützlichen Blöcke:

productionQueue.LinkTo(DataflowBlock.NullTarget<int>());