最近工作上遇到一个问题，就是有将近200万个CSV文件，每个CSV文件包含了成千上万条实验数据，CSV以一个不连续的整数值作为文件名，比如：1.CSV、2.CSV、3.CSV、5.CSV等等。另外又有200万个XML文件，每个XML文件的文件名与CSV的文件名一一对应，在这些XML文件中，定义了所对应的CSV实验数据文件的实验描述信息（比如实验名称、实验类型等等），也就是说，每个XML包含的是它所对应的CSV文件的元数据。现在的一个需求是，当软件中列出其中一部分（比如几千个或者几万个）CSV文件时，需要在每个文件名边上显示对应的实验名称。

咋一看这样的需求，感觉比较简单，当显示某个CSV文件时，直接找到对应的XML文件，解析XML得到名称就结束了。然而，问题是：

1. 这就需要软件本身自带这200万个XML文件，文件数量太大，如果压缩成ZIP，ZIP的尺寸也相对较大，在程序请求实验名称时还需要解压，性能极差
2. 解析XML本身需要损耗一定的性能，如果要显示成百上千个CSV对应的实验名称，那么需要对每个XML进行解析，性能也很不理想

在此，我介绍一种方法，通过预处理的方式，将所需信息提取成结构化的数据结构（Structured Data Structure），然后通过索引进行快速定位。

问题分析

虽然XML文件数量比较大，每个XML文件提供的信息也比较多，但是我们所需要的信息仅仅就是XML文件中的实验名称，因此，一个思路就是，首先对所有XML文件进行预处理，然后提取实验名称，并将其保存到另一个文件中。当需要根据CSV文件名获取实验名称时，就查询这个实验名称数据文件，然后显示对应的实验名称。这里的问题是，使用哪种格式来产生实验名称数据文件呢？我们又有几个选择：

1. 使用JSON，存储“CSV文件名<—>实验名称”的键值对，这样性能也不会很好，因为这样的键值对有200万个，解析JSON文件本身的CPU和IO负载会很高
2. 使用桌面数据库，比如SQLite，这样做需要应用程序内建一个SQLite的引擎，它本身存在CPU架构的问题（x86,x64），而且中间封了一层数据库访问操作，性能也不见得特别高
3. 自定义存储结构，这种做法比较灵活，但是需要自己实现，有一定的难度，出问题的几率也相对较大

综合分析，我们还是打算选择第三个方案，自己定义数据的存储结构。

假设CSV文件名是连续的，比如是从1.CSV、2.CSV一直到2000000.CSV，那么我们可以将CSV的文件名数值作为索引值，通过查表法找到对应的实验名称字符串即可。比如，在内存中有以下字符串数组：

假设CSV文件名为1535.CSV，那么我们只需要assayNames[1534]即可获得第1535个CSV（也就是1535.CSV）所对应的实验名称。这样做的效率是非常高的，它直接利用了数组的索引。然而，现实并不是那么美好：

1. 我们不可能把200万条数据全部放在一个数组内存中，这样做消耗内存会非常高
2. 原始CSV文件的文件名标号并不是连续的

解决问题一的方式比较直白：我们需要将数据放在磁盘中，然后按需访问；对于问题二，我们需要引入数据库实现中的一个概念：索引。

解决问题

假设每条实验名称数据被当成一条长度固定的记录存放在二进制文件中，但由于文件名中数值标识并不连续，因此，无法简单地通过文件名来推断数据记录的位置（也就是数组的下标值），比如：

对于1.csv、2.csv尚有规律可寻，实验名称数据记录在二进制文件中的位置，就是文件名数值减1，从4.csv开始，后面的位置值就与文件名没什么关系了。此时，我们需要有一个映射，来定义文件名中的数值与数据记录位置之间的关系。为此，我引入了另一个二进制文件，其中定义了200万条记录，每条记录仅占4个字节，每条记录（每4个字节）保存的是以该记录的偏移值作为文件名数值的CSV文件，所对应的实验名称数据记录在上述二进制文件中的记录位置。比如：

那么，假设CSV文件的文件名为4.csv，于是，可以首先找到索引文件中偏移值为4（也就是index=3）的记录位置值（也就是2），然后，在二进制文件中定位到索引值为2的记录，就是4.csv所对应的实验名称数据。

代码实现

我使用System.Runtime.InteropServices命名空间下的Marshal类和GCHandle类，配合System.IO命名空间下的BinaryReader、BinaryWriter类来实现结构化二进制文件的读取和写入。封装代码如下：

public static class BinaryFileHelper
{
    public static T ReadStruct<T>(BinaryReader binaryReader, int idx = 0)
        where T : struct
    {
        var buff = new byte[Marshal.SizeOf<T>()];

        if (binaryReader.BaseStream.CanSeek)
        {
            binaryReader.BaseStream.Seek(idx * buff.Length, SeekOrigin.Begin);
            binaryReader.BaseStream.Read(buff, 0, buff.Length);
        }

        var gcHandle = GCHandle.Alloc(buff, GCHandleType.Pinned);
        try
        {
            var result = Marshal.PtrToStructure<T>(gcHandle.AddrOfPinnedObject());
            return result;
        }
        finally
        {
            gcHandle.Free();
        }
    }

    public static void WriteStruct<T>(BinaryWriter binaryWriter, T item)
        where T : struct
    {
        var buff = new byte[Marshal.SizeOf<T>()];
        var gcHandle = GCHandle.Alloc(buff, GCHandleType.Pinned);
        try
        {
            Marshal.StructureToPtr<T>(item, gcHandle.AddrOfPinnedObject(), false);
            binaryWriter.Write(buff, 0, buff.Length);
        }
        finally
        {
            gcHandle.Free();
        }
    }
}

接下来，再写一个测试程序来测试结构化二进制文件的读取性能：

[StructLayout(LayoutKind.Explicit)]
public struct AssayNameStructuredIndex
{
    [FieldOffset(0)]
    [MarshalAs(UnmanagedType.U4, SizeConst = 4)]
    public int Index;
}

[StructLayout(LayoutKind.Explicit)]
public struct AssayNameStructuredRecord
{
    [FieldOffset(0)]
    [MarshalAs(UnmanagedType.ByValTStr, SizeConst = 256)]
    public string Name;
}

static void Main(string[] args)
{
    var stopwatch = new Stopwatch();
    using (var recordFileStream = new FileStream("assayNames.bin", FileMode.Open, FileAccess.Read))
    using (var indexFileStream = new FileStream("assayNames.idx", FileMode.Open, FileAccess.Read))
    using (var recordReader = new BinaryReader(recordFileStream))
    using (var indexReader = new BinaryReader(indexFileStream))
    {
        while (true)
        {
            Console.Write("请输入CSV文件名（直接回车退出程序）：");
            var line = Console.ReadLine();
            if (string.IsNullOrEmpty(line)) break;
            if (!int.TryParse(Path.GetFileNameWithoutExtension(line), out var identifier)) continue;
            stopwatch.Restart();
            var indexValue = BinaryFileHelper.ReadStruct<AssayNameStructuredIndex>(indexReader, identifier);
            if (indexValue.Index == -1)
            {
                Console.WriteLine($"数据文件中未包含{line}的记录。");
                Console.WriteLine();
                continue;
            }
            var assayNameValue = BinaryFileHelper.ReadStruct<AssayNameStructuredRecord>(recordReader, indexValue.Index);
            stopwatch.Stop();
            Console.WriteLine($"耗时：{stopwatch.ElapsedMilliseconds}毫秒，实验名称：{assayNameValue.Name}。");
            Console.WriteLine();
        }
    }
}

执行结果如下：

可以看到，无论CSV文件名中的数值是大还是小，从近200万条数据中读取实验名称信息的速度都是非常快的，基本上也就是零点几个毫秒，达到了预期的目标。

总结

所谓之结构化的数据，就是表示每条数据所占用的存储空间都是一致的，也就是每条记录所占用的字节数是相等的，这样才能非常容易地通过记录的索引值以及每条记录的大小来计算位置偏移量，从而快速读取数据。这是一种空间换时间的方案，一个明显的问题是，需要根据实际数据来合理选择每条记录所占用的存储空间：如果太大，那么200多万条记录累积起来，会占用大量存储空间，造成空间浪费；如果太小，又会导致某些数据无法正确存储，造成信息丢失。因此，本文介绍的方案还是需要根据实际情况进行斟酌，选择合理的记录存储结构。