C# 多线程 09-使用异步 I/O 01-异步地使用文件

🏷️ 《C# 多线程》

示例代码

/// <summary>
/// 异步地使用文件
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    var t = ProcessAsynchronousIO();
    t.GetAwaiter().GetResult();

    Console.ReadLine();
}

const int BUFFER_SIZE = 4096;

static async Task ProcessAsynchronousIO()
{
    // 使用 FileStream 创建文件
    using (var stream = new FileStream("test1.txt", FileMode.Create, FileAccess.ReadWrite, FileShare.None, BUFFER_SIZE))
    {
        Console.WriteLine($"1. Uses I/O Threads: {stream.IsAsync}");

        byte[] buffer = Encoding.UTF8.GetBytes(CreateFileContent());
        // 将异步编程模型 API 转换成任务
        var writeTask = Task.Factory.FromAsync(stream.BeginWrite, stream.EndWrite, buffer, 0, buffer.Length, null);
        await writeTask;
    }

    // 使用 FileStream 创建文件（提供了 FileOptions.Asynchronous 参数）
    // 只有提供了提供了 FileOptions.Asynchronous 选项，才能对 FileStream 类使用异步 IO
    using (var stream = new FileStream("test2.txt", FileMode.Create, FileAccess.ReadWrite, FileShare.None, BUFFER_SIZE, FileOptions.Asynchronous))
    {
        Console.WriteLine($"2. Uses I/O Threads: {stream.IsAsync}");

        byte[] buffer = Encoding.UTF8.GetBytes(CreateFileContent());
        // 将异步编程模型 API 转换成任务
        var writeTask = Task.Factory.FromAsync(stream.BeginWrite, stream.EndWrite, buffer, 0, buffer.Length, null);
        await writeTask;
    }

    // 使用 File.Create（提供了 FileOptions.Asynchronous 参数）和 StreamWriter 创建和写入文件
    using (var stream = File.Create("test3.txt", BUFFER_SIZE, FileOptions.Asynchronous))
    using (var sw = new StreamWriter(stream))
    {
        Console.WriteLine($"3. Uses I/O Threads: {stream.IsAsync}");
        // 异步写入流
        await sw.WriteAsync(CreateFileContent());
    }

    // 仅使用 StreamWriter 创建文件
    using (var sw = new StreamWriter("test4.txt", true))
    {
        Console.WriteLine($"4. Uses I/O Threads: {((FileStream)sw.BaseStream).IsAsync}");

        // 异步写入流（因为没有提供 FileOptions.Asynchronous 参数，Stream 其实并没有使用异步 I/O）
        await sw.WriteAsync(CreateFileContent());
    }

    Console.WriteLine("Starting parsing files in parallel");
    var readTasks = new Task<long>[4];
    for (int i = 0; i < 4; i++)
    {
        string fileName = $"test{i + 1}.txt";
        // 异步读取文件并 Sum
        readTasks[i] = SumFileContent(fileName);
    }
    // 等待所有异步 Task 完成，并获取返回值数组
    long[] sums = await Task.WhenAll(readTasks);
    Console.WriteLine($"Sum is all files: {sums.Sum()}");

    Console.WriteLine("Deleting files");
    Task[] deleteTasks = new Task[4];
    for (int i = 0; i < 4; i++)
    {
        string filename = $"test{i + 1}.txt";
        // 异步删除文件
        deleteTasks[i] = SimulateAsynchronousDelete(filename);
    }
    // 等待所有异步删除操作结束
    await Task.WhenAll(deleteTasks);
    Console.WriteLine("Deleting complete.");
}

/// <summary>
/// 异步删除文件
/// </summary>
/// <param name="filename"></param>
/// <returns></returns>
static Task SimulateAsynchronousDelete(string filename)
{
    // 使用 Task.Run 模拟异步删除
    return Task.Run(() => File.Delete(filename));
}

/// <summary>
/// 异步统计文件中随机数的合计值
/// </summary>
/// <param name="fileName"></param>
/// <returns></returns>
static async Task<long> SumFileContent(string fileName)
{
    // 使用 FileStream 和 StreamReader 异步读取文件
    using (var stream = new FileStream(fileName, FileMode.Open, FileAccess.Read, FileShare.None, BUFFER_SIZE, FileOptions.Asynchronous))
    using (var sr = new StreamReader(stream))
    {
        long sum = 0;
        while (sr.Peek() > -1)
        {
            string line = await sr.ReadLineAsync();
            sum += long.Parse(line);
        }
        return sum;
    }
}

/// <summary>
/// 创建随机的文件内容
/// </summary>
/// <returns></returns>
static string CreateFileContent()
{
    var sb = new StringBuilder();
    for (int i = 0; i < 100000; i++)
    {
        sb.Append($"{ new Random(DateTime.Now.Millisecond).Next(0, 99999)}");
        sb.AppendLine();
    }
    return sb.ToString();
}

运行结果

1. Uses I/O Threads: False
2. Uses I/O Threads: True
3. Uses I/O Threads: True
4. Uses I/O Threads: False
Starting parsing files in parallel
Sum is all files: 19850361639
Deleting files
Deleting complete.