如何在.NET中通过非阻塞IO并发读取文件行？

首先要明确：你的BiggestLineConcurrent方法报错，是因为StreamReader的异步API不支持同时发起多个未完成的异步操作——它的内部流状态是单线程设计的，前一个ReadLineAsync还没完成时，再次调用会导致流资源冲突，抛出InvalidOperationException。

为什么文件行读取没法直接“并发”？ #

文件系统本身是顺序访问的，文件指针只能单向移动，你不可能同时在文件的不同位置读取不同行（除非将文件拆分为独立的块）。非阻塞IO的核心是让线程在等待IO完成时释放出来处理其他任务，而不是让同一个文件的读取操作并行执行。

正确的异步非阻塞处理方式 #

如果你想避免线程阻塞、提升效率，顺序的异步读取已经是非阻塞的了，它不会占用线程等待IO完成。如果需要并行处理读取到的行（而不是并行读取），可以在读取后用并行LINQ或Task.WhenAll处理结果：

static async Task<string> BiggestLineAsyncProcess(String filename)
{
    List<string> lines = new List<string>();
    using (StreamReader reader = new StreamReader(filename))
    {
        string line;
        while ((line = await reader.ReadLineAsync()) != null)
        {
            lines.Add(line);
        }
    }

    // 并行处理所有行找最长的
    return lines.AsParallel()
                .Where(l => l != null)
                .OrderByDescending(l => l.Length)
                .FirstOrDefault() ?? string.Empty;
}

真正的文件并行读取（分块方式） #

如果确实需要并行读取超大文件的不同部分，可以将文件分成多个块，每个块单独打开FileStream并设置读取起始位置，借助.NET的IOCP完成端口实现非阻塞并发。注意要处理跨块的不完整行：

static async Task<string> BiggestLineParallelFile(String filename)
{
    long fileSize = new FileInfo(filename).Length;
    int chunkCount = Environment.ProcessorCount;
    long chunkSize = fileSize / chunkCount;
    List<Task<string>> chunkTasks = new List<Task<string>>();

    for (int i = 0; i < chunkCount; i++)
    {
        long startPos = i * chunkSize;
        // 最后一块包含剩余所有内容
        long endPos = i == chunkCount - 1 ? fileSize : startPos + chunkSize;
        chunkTasks.Add(ProcessFileChunk(filename, startPos, endPos));
    }

    string[] chunkResults = await Task.WhenAll(chunkTasks);
    return chunkResults.OrderByDescending(s => s.Length).FirstOrDefault() ?? string.Empty;
}

static async Task<string> ProcessFileChunk(string filename, long startPos, long endPos)
{
    string longestLine = string.Empty;
    using (FileStream stream = new FileStream(filename, FileMode.Open, FileAccess.Read, FileShare.Read, 4096, FileOptions.Asynchronous))
    {
        stream.Position = startPos;
        using (StreamReader reader = new StreamReader(stream))
        {
            // 如果不是文件开头，先跳过第一行（可能是不完整的行）
            if (startPos > 0)
                await reader.ReadLineAsync();

            string line;
            while ((line = await reader.ReadLineAsync()) != null && stream.Position <= endPos)
            {
                if (line.Length > longestLine.Length)
                    longestLine = line;
            }
        }
    }
    return longestLine;
}

结论 #

1. 无需显式创建线程/线程池，用异步IO就能实现非阻塞的文件读取，但单文件的行读取本身无法真正并发，因为流是顺序的。
2. 如果要并行处理行，可先异步读取所有行，再用PLINQ或Task.WhenAll并行处理结果。
3. 超大文件的并行读取需要分块处理，每个块单独打开异步流，这种方式依赖.NET底层机制实现非阻塞并发，同样无需手动创建线程。

内容的提问来源于stack exchange，提问作者Diogo Rodrigues