Многопоточная обработка последовательных данных

Question

Вопрос по мотивам MPEG TS обработка пакетов в нескольких потоках Попытался абстрагироваться от специфики вопроса. Есть абстрактная фабрика:

 public abstract class AbstractFactory
{        
    public abstract void PushInt(int num);
}

Есть следующие конкретные фабрики:

public class Factory_9 : AbstractFactory
{
    List<int> Ints = new List<int>();
    public override void PushInt(int num)
    {
        Ints.Add(num);
        Thread.Sleep(50);
    }
}
public class Factory_8 : AbstractFactory
{
    List<int> Ints = new List<int>();
    public override void PushInt(int num)
    {
    Ints.Add(num);
    Thread.Sleep(50);
}

}

итд до public class Factory_5

есть главный парсер:

public class MainParser
{
    private readonly Lazy<Factory_9> factory_9 = new Lazy<Factory_9>();
    private readonly Lazy<Factory_8> factory_8 = new Lazy<Factory_8>();
    private readonly Lazy<Factory_7> factory_7 = new Lazy<Factory_7>();
    private readonly Lazy<Factory_5> factory_5 = new Lazy<Factory_5>();
    private readonly Lazy<Factory_6> factory_6 = new Lazy<Factory_6>();
public Factory_9 _9 =&gt; factory_9.Value;
public Factory_8 _8 =&gt; factory_8.Value;
public Factory_7 _7 =&gt; factory_7.Value;
public Factory_5 _5 =&gt; factory_5.Value;
public Factory_6 _6 =&gt; factory_6.Value;

public void PushIntArray(int[] array)
{
    foreach( var num in array)
    {
        int firstDigit = getFirstDigit(num);

        if (firstDigit == 9)
        {
            _9.PushInt(num);
        }
        else if(firstDigit == 8)
        {
            _8.PushInt(num);
        }
        else if(firstDigit == 7)
        {
            _7.PushInt(num);
        }
        else if(firstDigit == 6)
        {
            _6.PushInt(num);
        }
        else if(firstDigit == 5)
        {
            _5.PushInt(num);
        }
        else
        {

        }
    }
}

private int getFirstDigit(int n)
{

    while (n &gt;= 10)
        n /= 10;


    return n;
}

}

Фактически в данной задаче входные данные представляют собой рандомный массив int, но по легенде все числа, которые начинаются с одной и той же цифры взаимосвязаны (как пакеты данных в MPEG TS) Парсер раскидывает данные по фабрикам в зависимости от первой цифры числа. Для каждой цифры должна создаваться только одна фабрика, так как в приходящей последовательности эти числа взаимосвязаны. В текущей реализации в List<int> каждой фабрики должны попасть только числа с первой цифрой из названия фабрики. То есть Factory_9 только числа на 9, Factory_8 на 8 и тд. и только в той же последовательности что и в исходном массиве. "Проблема" в том, что фабрики 9,8,7 имеют задержку по 50 мс на добавление числа в массив. И если запустить программу, то время выполнения получается примерно 15900-16100 мс. На каждую фабрику приходится примерно по ~5000 мс времени выполнения. То есть если мы сделаем параллельную работу этих трёх фабрик, то полное время работы приложения будет примерно 5000-6000 мс. Иными словами сократится в 3-раза для этого примера. Как распараллелить эту задачу правильно?

Answer 1

С использованием BlockingCollection класс MainParser получился следующего вида:

public class MainParser
{
    private readonly Lazy<Factory_9> factory_9 = new Lazy<Factory_9>();
    private readonly Lazy<Factory_8> factory_8 = new Lazy<Factory_8>();
    private readonly Lazy<Factory_7> factory_7 = new Lazy<Factory_7>();
    private readonly Lazy<Factory_5> factory_5 = new Lazy<Factory_5>();
    private readonly Lazy<Factory_6> factory_6 = new Lazy<Factory_6>();
private BlockingCollection&lt;int&gt; _9_collection = new BlockingCollection&lt;int&gt;();
private BlockingCollection&lt;int&gt; _8_collection = new BlockingCollection&lt;int&gt;();
private BlockingCollection&lt;int&gt; _7_collection = new BlockingCollection&lt;int&gt;();

Task task_9;
Task task_8;
Task task_7;

public Factory_9 _9 =&gt; factory_9.Value;
public Factory_8 _8 =&gt; factory_8.Value;
public Factory_7 _7 =&gt; factory_7.Value;
public Factory_5 _5 =&gt; factory_5.Value;
public Factory_6 _6 =&gt; factory_6.Value;

public MainParser()
{
    task_9 = Task.Run(Parse_9);
    task_8 = Task.Run(Parse_8);
    task_7 = Task.Run(Parse_7);
}

public void PushIntArray(int[] array)
{
    foreach( var num in array)
    {
        int firstDigit = getFirstDigit(num);

        if (firstDigit == 9)
        {
            _9_collection.Add(num);
        }
        else if(firstDigit == 8)
        {
            _8_collection.Add(num);
        }
        else if(firstDigit == 7)
        {
            _7_collection.Add(num);
        }
        else if(firstDigit == 6)
        {
            _6.PushInt(num);
        }
        else if(firstDigit == 5)
        {
            _5.PushInt(num);
        }
        else
        {

        }
    }

    _9_collection.CompleteAdding();
    _8_collection.CompleteAdding();
    _7_collection.CompleteAdding();

    Task.WaitAll(task_9, task_8, task_7);
}

private int getFirstDigit(int n)
{

    while (n &gt;= 10)
        n /= 10;


    return n;
}

private void Parse_9()
{
    int i;
    while (!_9_collection.IsCompleted)
    {
        if(_9_collection.TryTake(out i))
        {
            _9.PushInt(i);
        }
    }
}
private void Parse_8()
{
    int i;
    while (!_8_collection.IsCompleted)
    {
        if (_8_collection.TryTake(out i))
        {
            _8.PushInt(i);
        }
    }
}
private void Parse_7()
{
    int i;
    while (!_7_collection.IsCompleted)
    {
        if (_7_collection.TryTake(out i))
        {
            _7.PushInt(i);
        }
    }
}

}

Время выполнения 5683 мс в среднем. Прирост производительности в 3 раза. паттерн producer/consumer.