c# - 多线程计算从多个文件加载数据。如何在 C# 中使用和同步 StreamReader？

Question

如果我的问题有任何不清楚的地方，请告诉我如何让它变得更好。我是这个论坛的新手，如有任何错误，请提前致歉。

我正在对多种金融工具进行计算。对于它们中的每一个，我都有两个单独的文件 - 一个包含给定市场上的交易记录，第二个包含订单簿的当前状态(bidask 大小和价格)。由于文件很大，每次计算我都需要加载一行数据文件。出于同样的原因，将文件合并在一起也不是一个好主意。

我的问题是，如何使用两个 StreamReader(或任何其他方式？)来执行以下计算:

1. 加载两行
2. 根据时间戳决定哪个先出现
3. 对第一个执行计算并将第二个保存到临时
4. 载入另一行计算文件
5. 与温度比较(第2步)
6. 重复

对于一个文件来说，这是非常微不足道的。计算如下:

string[] ListOfRawData = Directory.GetFiles(...);
Parallel.ForEach
        (
        ListOfRawData,
        new ParallelOptions { MaxDegreeOfParallelism = 8 },
        (FileName) =>
        {
            SupportMethods Support = new SupportMethods(FileName);
            List<Calculation> CalculationInstances = new List<Calculation>();
            using (StreamReader DataFile = new StreamReader(FileName))
            {
                while (!DataFile.EndOfStream)
                {
                    // Determine whether the line is new trade or an orderbook update
                    Support.TradeOrUpdate(DataFile);

                    switch (Support.IsTrade)
                    {
                        case false:
                            foreach (Calculation ActualCalculation in CalculationInstances)
                            {
                                ActualCalculation.DOMUpdateMethod(DMUpdate, Support);
                            }
                            break;
                        case true:
                            foreach (Calculation ActualCalculation in CalculationInstances)
                            {
                                ActualCalculation.TradeUpdateMethod(TradeUpdate, Support);
                            }
                            break;
                        default:
                            break;
                    }
                }
            }
      }
      );

我想执行相同的计算，但是现在从两个单独的文件加载。

在我的搜索过程中，我读到使用两个流阅读器需要两个单独的线程 - 是这样吗？这会不会有问题，因为我使用 Parallel.ForEach 在多个线程上同时为 8 种不同的仪器运行上述计算？

这两个文件都由时间戳标识如下:

交易文件:

02-JAN-2007,08:21:13.173,+1,Trade,40.6,1000,40.6800 (...)
02-JAN-2007,08:21:16.379,+1,Trade,40.6,130,40.6800 (...)

订单簿文件:

02-JAN-2007,08:21:13.274,+1,Market Depth,40.6,4000,1,40.7,4731,3, (...)
02-JAN-2007,08:21:17.345,+1,Market Depth,40.6,3870,1,40.7,4731,3, (...)

任何想法或链接将不胜感激。

编辑: 我感兴趣的代码部分在“using Streamreader”和“swich”语句之间。 parallel.foreach 可能令人困惑，我添加它只是为了完整性。

Answer 1

您似乎在寻找合并联接算法。它需要两个已排序的序列，并将它们合并为一个常量内存。如果你不想自己做，你可以在网上找到这样的算法。

代码看起来像这样:

IEnumerable<...> src1 = GetSource1();
IEnumerable<...> src2 = GetSource2();
IEnumerable<...> merged = Merge(src1, src2, x => x.GetTimeStamp(), x => x.GetTimeStamp());
merged
 .AsParallel()
 .ForEach(x => PerformUpdate(x));

这使得读取文件成为单线程并处理连接的结果成为多线程。