C# AspNetCore。数据处理的并行化。如何让它变得更好

您好，在处理 AspNetCore 2.2 上的 Web 服务的到达数据之前，需要执行一项任务。

该服务启动多达 100 个通过 POST 请求接收数据的并行设备。我不会描述设备如何处理这些数据。

在处理这些数据之前有一项任务要做。例如，在输入中，有一个字符串类型的字段“名称”，其值为“MyLongName”。工程师制定了如何转换此字符串的规则。

1. 如果超过限制修剪线
2. 在正确的位置插入子字符串
3. 更改案例
4. ...

那些。构建了一个动作管道来处理该类型。

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 并行（保持转换器调用的顺序）。您还可以使用 Parallel.ForEach 并行开始迭代 stringHandler 的数据本身。

那些。我们得到一个从 1 开始并行处理的输入数据列表。在这个数据单元的处理程序中，所有字符串类型都是并行处理的。字符串处理管道也是并行的（保留调用的顺序）。

我考虑过使用 MapReduce，但处理结束时不需要 Reduce（卷积），最后会有与输入相同的数据列表，只有转换后的数据。这种方式并行化处理是正常的，还是需要手动限制线程数，比如Parallel.ForEach。或者也许在某处使用 Task-i。

我将举一个具体的例子来分析并行操作设备100个。每个设备接受一个包含 50 个项目的列表。每个元素由 10 个 Handler 处理（需要更改 10 个 String 类型） 每个元素的处理管道由 5 个阶段组成。