发布于 由admin

C++ 异步 I/O (Asynchronous I/O)：系统层面的非阻塞操作

各位观众，各位朋友，各位未来的编程大神们，大家好！

今天咱们来聊聊C++里的异步I/O，这玩意儿听起来高大上，但其实就是让你的程序在等待数据的时候，别傻乎乎地在那儿杵着，而是可以先去干点别的，等数据来了再回来处理，大大提高效率。想象一下，你一边烧水一边写代码，水开了再回去泡茶，总比你一直盯着水壶看要强得多吧？这就是异步I/O的精髓。

什么是异步I/O？

首先，我们得明白同步I/O和异步I/O的区别。

  • 同步I/O: 你发起一个I/O操作，程序就得老老实实地等着，直到操作完成才能继续往下走。就像你排队买东西，必须等到轮到你，付完钱才能离开。

  • 异步I/O: 你发起一个I/O操作，然后就可以去做别的事情了，系统会在I/O操作完成后通知你，你再回来处理结果。就像你网购，下单后就可以去刷剧了，快递到了会通知你。

用表格来总结一下：

特性 同步I/O 异步I/O
等待方式 阻塞，必须等待完成 非阻塞，可以执行其他任务
效率 较低，浪费CPU时间 较高，提高CPU利用率
编程模型 简单，易于理解 复杂，需要处理回调

系统层面的非阻塞操作

异步I/O的核心在于“系统层面”，也就是说，这个非阻塞不是你自己在用户空间里实现的，而是操作系统内核提供的功能。内核会帮你处理I/O操作，并在完成后通知你。

C++ 中的异步 I/O 实现方式

C++本身并没有直接的异步I/O API，但是我们可以通过操作系统提供的API来实现。常见的方法有：

  1. POSIX AIO (Asynchronous I/O): 这是POSIX标准定义的一套异步I/O接口，在Linux和一些其他Unix-like系统上可以使用。

  2. Windows I/O Completion Ports: Windows提供的一套高效的异步I/O机制。

  3. 第三方库： 比如Boost.Asio、libuv等，它们封装了底层操作系统的异步I/O API，提供了更方便易用的接口。

接下来，我们分别看看这几种方式的实现和使用。

1. POSIX AIO

POSIX AIO 是一组函数，允许程序在不阻塞的情况下执行I/O操作。关键的函数包括：

  • aio_read(): 异步读取数据。
  • aio_write(): 异步写入数据。
  • aio_error(): 获取异步I/O操作的错误状态。
  • aio_return(): 获取异步I/O操作的返回值。
  • aio_cancel(): 取消一个异步I/O操作。
  • aio_suspend(): 等待一个或多个异步I/O操作完成。

示例代码 (Linux)

#include <iostream>
#include <aio.h>
#include <fcntl.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>

int main() {
    const char* filename = "test.txt";
    const int buffer_size = 1024;
    char buffer[buffer_size];

    // 打开文件
    int fd = open(filename, O_RDONLY);
    if (fd == -1) {
        std::cerr << "Error opening file: " << strerror(errno) << std::endl;
        return 1;
    }

    // 初始化 aio 控制块
    aiocb aio_control_block;
    memset(&aio_control_block, 0, sizeof(aiocb));
    aio_control_block.aio_fildes = fd;
    aio_control_block.aio_buf = buffer;
    aio_control_block.aio_nbytes = buffer_size;
    aio_control_block.aio_offset = 0;
    aio_control_block.aio_sigevent.sigev_notify = SIGEV_NONE; // 不使用信号通知

    // 发起异步读取
    int ret = aio_read(&aio_control_block);
    if (ret == -1) {
        std::cerr << "Error initiating asynchronous read: " << strerror(errno) << std::endl;
        close(fd);
        return 1;
    }

    // 在这里可以做其他事情...
    std::cout << "Asynchronous read initiated. Doing other stuff..." << std::endl;

    // 等待 I/O 操作完成
    while (aio_error(&aio_control_block) == EINPROGRESS) {
        // 可以选择睡眠一段时间，减少CPU占用
        usleep(1000); // 1毫秒
    }

    // 检查 I/O 操作是否成功
    ret = aio_return(&aio_control_block);
    if (ret == -1) {
        std::cerr << "Error during asynchronous read: " << strerror(errno) << std::endl;
        close(fd);
        return 1;
    }

    // 读取的数据长度
    ssize_t bytes_read = ret;

    // 输出读取的数据
    std::cout << "Read " << bytes_read << " bytes:n" << buffer << std::endl;

    // 关闭文件
    close(fd);

    return 0;
}

代码解释：

  1. 包含头文件: 包含了必要的头文件，如 <aio.h> (异步I/O), <fcntl.h> (文件控制), <errno.h> (错误处理), <string.h> (字符串操作), <unistd.h> (POSIX系统调用)。

  2. 打开文件: 使用 open() 函数以只读模式打开文件 "test.txt"。如果打开失败，则打印错误信息并退出。

  3. 初始化 aiocb 结构体: 创建一个 aiocb (异步I/O控制块) 结构体，并使用 memset() 初始化为零。

    • aio_fildes: 设置为文件描述符 fd，指定要操作的文件。
    • aio_buf: 设置为缓冲区 buffer，用于存储读取的数据。
    • aio_nbytes: 设置为缓冲区的大小 buffer_size，指定要读取的字节数。
    • aio_offset: 设置为 0，表示从文件开头开始读取。
    • aio_sigevent.sigev_notify: 设置为 SIGEV_NONE，表示不使用信号来通知I/O完成。

  4. 发起异步读取: 使用 aio_read() 函数发起异步读取操作，将 aiocb 结构体作为参数传递。如果函数返回 -1，表示发起异步读取失败，打印错误信息并关闭文件后退出。

  5. 执行其他任务: 在异步读取发起后，可以执行其他任务，这里简单地打印一条消息。

  6. 等待I/O操作完成: 使用 aio_error() 函数轮询检查异步I/O操作的状态。当 aio_error() 返回 EINPROGRESS 时，表示操作仍在进行中。可以使用 usleep() 函数睡眠一段时间，以减少CPU占用。

  7. 检查I/O操作结果: 使用 aio_return() 函数获取异步I/O操作的结果。如果函数返回 -1，表示操作失败，打印错误信息并关闭文件后退出。

  8. 处理读取的数据: 如果异步I/O操作成功完成，aio_return() 函数返回读取的字节数。然后，打印读取的数据。

  9. 关闭文件: 使用 close() 函数关闭文件描述符。

注意事项：

  • POSIX AIO 可能不是在所有系统上都真正实现异步，有些系统可能只是模拟异步，性能会受到影响。
  • 你需要确保提供的缓冲区在I/O操作完成之前有效，不能被释放或修改。
  • 错误处理非常重要，你需要检查 aio_error() 和 aio_return() 的返回值，以确保I/O操作成功完成。

2. Windows I/O Completion Ports

Windows I/O Completion Ports (IOCP) 是一种高效的异步I/O机制，特别适合处理大量并发的I/O操作。

基本步骤：

  1. 创建 I/O Completion Port: 使用 CreateIoCompletionPort() 函数创建一个IOCP。
  2. 将文件句柄关联到 I/O Completion Port: 使用 CreateIoCompletionPort() 函数将文件句柄与IOCP关联起来。
  3. 发起异步 I/O 操作: 使用 ReadFile() 或 WriteFile() 函数发起异步I/O操作，并传入一个 OVERLAPPED 结构体。
  4. 等待 I/O 完成: 使用 GetQueuedCompletionStatus() 函数等待I/O操作完成，并获取结果。

示例代码 (Windows)

#include <iostream>
#include <windows.h>

int main() {
    const char* filename = "test.txt";
    const int buffer_size = 1024;
    char buffer[buffer_size];
    OVERLAPPED overlapped;
    HANDLE hFile = INVALID_HANDLE_VALUE;
    HANDLE hCompletionPort = NULL;

    // 1. 创建 I/O Completion Port
    hCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
    if (hCompletionPort == NULL) {
        std::cerr << "CreateIoCompletionPort failed: " << GetLastError() << std::endl;
        return 1;
    }

    // 2. 打开文件 (必须以 OVERLAPPED 方式打开)
    hFile = CreateFileA(filename, GENERIC_READ, 0, NULL, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED, NULL);
    if (hFile == INVALID_HANDLE_VALUE) {
        std::cerr << "CreateFile failed: " << GetLastError() << std::endl;
        CloseHandle(hCompletionPort);
        return 1;
    }

    // 3. 将文件句柄关联到 I/O Completion Port
    if (CreateIoCompletionPort(hFile, hCompletionPort, (ULONG_PTR)hFile, 0) == NULL) {
        std::cerr << "CreateIoCompletionPort (associate file handle) failed: " << GetLastError() << std::endl;
        CloseHandle(hFile);
        CloseHandle(hCompletionPort);
        return 1;
    }

    // 4. 初始化 OVERLAPPED 结构体
    memset(&overlapped, 0, sizeof(overlapped));
    overlapped.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL); // Manual-reset event

    // 5. 发起异步读取
    DWORD bytesRead;
    if (!ReadFile(hFile, buffer, buffer_size, &bytesRead, &overlapped)) {
        if (GetLastError() != ERROR_IO_PENDING) {
            std::cerr << "ReadFile failed: " << GetLastError() << std::endl;
            CloseHandle(hFile);
            CloseHandle(hCompletionPort);
            CloseHandle(overlapped.hEvent);
            return 1;
        }
        // ERROR_IO_PENDING 表示异步操作正在进行中
    }

    // 6. 在这里可以做其他事情...
    std::cout << "Asynchronous read initiated. Doing other stuff..." << std::endl;

    // 7. 等待 I/O 完成
    DWORD bytesTransferred;
    ULONG_PTR completionKey;
    LPOVERLAPPED lpOverlapped;

    BOOL bRet = GetQueuedCompletionStatus(hCompletionPort, &bytesTransferred, &completionKey, &lpOverlapped, INFINITE);
    if (!bRet) {
        std::cerr << "GetQueuedCompletionStatus failed: " << GetLastError() << std::endl;
        CloseHandle(hFile);
        CloseHandle(hCompletionPort);
        CloseHandle(overlapped.hEvent);
        return 1;
    }

    // 8. 检查 I/O 操作是否成功
    if (lpOverlapped != &overlapped) {
        std::cerr << "Unexpected OVERLAPPED structure returned." << std::endl;
        CloseHandle(hFile);
        CloseHandle(hCompletionPort);
        CloseHandle(overlapped.hEvent);
        return 1;
    }

    // 9. 输出读取的数据
    std::cout << "Read " << bytesTransferred << " bytes:n" << buffer << std::endl;

    // 10. 清理资源
    CloseHandle(hFile);
    CloseHandle(hCompletionPort);
    CloseHandle(overlapped.hEvent);

    return 0;
}

代码解释：

  1. 创建 I/O Completion Port: 使用 CreateIoCompletionPort() 创建一个IOCP。如果创建失败，则输出错误信息并退出。

  2. 打开文件 (OVERLAPPED): 使用 CreateFileA() 函数以 OVERLAPPED 方式打开文件。OVERLAPPED 标志是使用 IOCP 的关键。如果打开失败，则输出错误信息并关闭IOCP后退出。

  3. 关联文件句柄: 再次使用 CreateIoCompletionPort() 将文件句柄与IOCP关联起来。completionKey 参数可以用来区分不同的I/O操作。

  4. 初始化 OVERLAPPED 结构体: 初始化 OVERLAPPED 结构体。hEvent 成员用于在I/O操作完成时发出信号，这里创建了一个 manual-reset event。

  5. 发起异步读取: 使用 ReadFile() 函数发起异步读取操作。如果函数返回 FALSE，并且 GetLastError() 返回 ERROR_IO_PENDING，则表示异步操作正在进行中。

  6. 执行其他任务: 在异步读取发起后，可以执行其他任务。

  7. 等待 I/O 完成: 使用 GetQueuedCompletionStatus() 函数等待I/O操作完成。该函数会阻塞，直到有I/O操作完成。

  8. 检查 I/O 操作结果: 检查 GetQueuedCompletionStatus() 返回的 lpOverlapped 是否与之前传入的 overlapped 结构体相同。

  9. 输出读取的数据: 如果异步I/O操作成功完成，则输出读取的数据。

  10. 清理资源: 关闭文件句柄、IOCP句柄和事件句柄。

注意事项：

  • 必须以 FILE_FLAG_OVERLAPPED 标志打开文件，才能使用IOCP。
  • OVERLAPPED 结构体是异步I/O的关键，它用于跟踪I/O操作的状态。
  • GetQueuedCompletionStatus() 函数会阻塞，直到有I/O操作完成。可以使用超时参数来避免无限期阻塞。
  • IOCP 适用于高并发的I/O操作，但编程模型相对复杂。

3. 第三方库：Boost.Asio

Boost.Asio 是一个跨平台的C++库，提供了网络和底层I/O的抽象。它简化了异步I/O的编程，并提供了更高级的接口。

示例代码 (Boost.Asio)

#include <iostream>
#include <boost/asio.hpp>
#include <boost/asio/ts/buffer.hpp>
#include <boost/asio/ts/internet.hpp>

int main() {
    boost::asio::io_context io_context;
    boost::asio::ip::tcp::acceptor acceptor(io_context, {boost::asio::ip::tcp::v4(), 55555}); // 监听 55555 端口
    boost::asio::ip::tcp::socket socket(io_context);

    acceptor.async_accept(socket, [&](boost::system::error_code ec) { // 异步接受连接
        if (!ec) {
            std::cout << "Connection accepted!" << std::endl;
            std::string message = "Hello from server!n";
            boost::asio::async_write(socket, boost::asio::buffer(message),
                [&](boost::system::error_code ec, std::size_t bytes_transferred) { // 异步写入数据
                    if (!ec) {
                        std::cout << "Message sent: " << message;
                        socket.close();
                    } else {
                        std::cerr << "Write error: " << ec.message() << std::endl;
                    }
                });
        } else {
            std::cerr << "Accept error: " << ec.message() << std::endl;
        }
    });

    io_context.run(); // 运行 io_context，处理异步事件

    return 0;
}

代码解释：

  1. 包含头文件: 包含了 Boost.Asio 的头文件。

  2. 创建 io_context: io_context 是 Asio 的核心组件，用于管理异步操作。

  3. 创建 acceptor: acceptor 用于监听指定端口的连接请求。

  4. 创建 socket: socket 用于与客户端进行通信。

  5. 异步接受连接: 使用 async_accept() 函数异步接受连接。当有客户端连接时，会调用 lambda 表达式。

  6. 异步写入数据: 在连接建立后，使用 async_write() 函数异步写入数据。当数据发送完成后，会调用另一个 lambda 表达式。

  7. 运行 io_context: io_context.run() 函数会阻塞，直到所有异步操作完成。它会不断地检查是否有事件发生，并调用相应的回调函数。

注意事项：

  • Boost.Asio 提供了更高级的抽象，简化了异步I/O的编程。
  • 你需要理解 Asio 的基本概念，如 io_context、socket、buffer 等。
  • Boost.Asio 是一个跨平台的库，可以在不同的操作系统上使用。

总结

异步 I/O 是一种强大的技术，可以显著提高程序的性能和响应速度。但是，它也增加了编程的复杂性。你需要根据你的具体需求和应用场景，选择合适的异步I/O实现方式。

记住，掌握异步I/O不是一蹴而就的，需要不断地学习和实践。多写代码，多查资料，你也能成为异步I/O的高手！

好了，今天的分享就到这里，希望对大家有所帮助。下次再见！

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注