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

Вопрос по мотивам 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 => factory_9.Value;
    public Factory_8 _8 => factory_8.Value;
    public Factory_7 _7 => factory_7.Value;
    public Factory_5 _5 => factory_5.Value;
    public Factory_6 _6 => 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 >= 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-раза для этого примера. Как распараллелить эту задачу правильно?


Ответы (1 шт):

Автор решения: Deim

С использованием 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<int> _9_collection = new BlockingCollection<int>();
    private BlockingCollection<int> _8_collection = new BlockingCollection<int>();
    private BlockingCollection<int> _7_collection = new BlockingCollection<int>();

    Task task_9;
    Task task_8;
    Task task_7;

    public Factory_9 _9 => factory_9.Value;
    public Factory_8 _8 => factory_8.Value;
    public Factory_7 _7 => factory_7.Value;
    public Factory_5 _5 => factory_5.Value;
    public Factory_6 _6 => 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 >= 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.

→ Ссылка