Здравствуйте Стоит задача сделать пред обработку поступающих данных для Web сервиса на AspNetCore 2.2.
Сервис запускает до 100 параллельно работающих устройств, которые получают данные через POST запрос. Что делает устройство с этими данными описывать не буду.
Появилась задача сделать пред обработку этих данных. Например, во входных данных, есть поле "Name" типа String со значением "MyLongName" . Инженеры задают правила, как преобразовать эту строку.
- Обрезать строку если больше лимита
- Вставить подстроки в нужные места
- Сменить регистр
- ...
Т.е. на обработку типа выстраивается конвейер из действий.
public class MiddleWareInData<TIn> : IDisposable
{
public List<StringHandlerMiddleWare> StringHandlers { get; }
public List<DateTimeHandlerMiddleWare> DateTimeHandlers { get; }
private void HandleInvoke(IEnumerable<TIn> datas)
{
foreach (var data in datas)
{
//ОБРАБОТЧИКИ String
Parallel.ForEach(StringHandlers, (stringHandler) =>
{
//Псевдо код обработки !!!
var str = "Начальная строка"; //Найденное свойство в типе через рефлексию
var res = stringHandler.Convert(str); //Преобразование
str = res; //перезаписали занчение свойства
});
//ОБРАБОТЧИКИ DateTime
//foreach (var dateTimeHandler in DateTimeHandlers)
//{
//}
}
}
}
public class StringHandlerMiddleWare : BaseHandlerMiddleWare<string>
{
public StringHandlerMiddleWare(StringHandlerMiddleWareOption option)
{
PropName = option.PropName;
if (option.InseartStringConverterOption != null)
{
Converters.Add(new InseartStringConverter(option.InseartStringConverterOption));
}
if (option.LimitStringConverterOption != null)
{
Converters.Add(new LimitStringComverter(option.LimitStringConverterOption));
}
if (option.ReplaceEmptyStringConverterOption != null)
{
Converters.Add(new ReplaceEmptyStringConverter(option.ReplaceEmptyStringConverterOption));
}
}
}
public abstract class BaseHandlerMiddleWare<T>
{
public string PropName { get; protected set; }
protected readonly List<IConverterMiddleWare<T>> Converters = new List<IConverterMiddleWare<T>>();
public virtual T Convert(T inProp)
{
var processedPrallel = Converters
.AsParallel()
.AsOrdered();
foreach (var converter in processedPrallel)
{
inProp = converter.Convert(inProp);
}
return inProp;
}
}
В классе MiddleWareInData данные поступают в метод HandleInvoke, в котором последовательно по 1 начинают обрабатываться. Я решил распараллелить обработку. Самое простое применить Parallel.ForEach и распараллелеить обработку по типам (параллельно обрабатываются все типы String для единицы данных) Внутри обработки типа String, запускается конвейер обработки типа, который тоже можно запустить параллельно (С СОХРАНЕНИЕМ ПОРЯДКА ВЫЗОВОВ КОНВЕРТЕРОВ) с помощью PLINQ. Еще можно параллельно запустить перебор самих данных для stringHandler также с помощью Parallel.ForEach.
Т.е. мы получим список входных данных которые начинают параллельно обрабатываться по 1. Внутри обработчика этой единицы данных, все строковые типы обрабатываются параллельно. Конвейер обработки строкового типа также параллелен (с сохранением порядка вызовов).
Думал над применением MapReduce, но Reduce (свертка) в конце обработки не нужна, в конце будет тот же список данных что и на входе, только с преобразованными данными. Нормально ли так параллелить обработку или надо ограничивать кол-во потоков в ручную например для Parallel.ForEach. Или может где-то использовать Task-и.
Возьму конкретный пример для анализа Параллельно работающих устройств 100 штук. Каждое устройство принимает список из 50 элементов. Каждый элемент обрабатыватся 10 Обработчиками (10 типов String нужно поменять) Конвейер обработки каждого элемента состоит из 5 этапов.