本文介绍一个基于 io_uring 的生产级 C++ 异步 IO 库实现，涵盖内核特性、业务封装及性能优化案例。
一、io_uring：异步 IO
1.1 传统 IO 模型的困境
在深入 io_uring 之前，我们先回顾一下传统 IO 模型的问题。
**同步阻塞 IO（Blocking IO）**是最简单直观的模型。当调用 read() 或 write() 时，线程会一直阻塞，直到数据准备就绪或写入完成。在单线程程序中，这意味着整个程序停滞；在多线程程序中，虽然可以通过创建更多线程来提高并发度，但每个线程的上下文切换开销（通常 1-5 微秒）和内存开销（每个线程的栈空间至少几 MB）会迅速积累，当线程数达到数千时，系统开销将变得不可接受。
**同步非阻塞 IO（Non-blocking IO）+ 多路复用（epoll/select）**是 Linux 服务器开发的经典方案。通过 epoll_wait() 监听多个文件描述符的就绪事件，单个线程可以同时处理数千个连接。然而，这个模型对于文件 IO 并不理想，因为：
- 文件描述符在
epoll中总是就绪的（无法真正监听磁盘 IO 完成） - 实际的
read()/write()调用仍然会阻塞线程，等待数据从磁盘加载到页缓存
**POSIX 异步 IO（aio）**试图解决这个问题，但其实现存在诸多缺陷：
- 不同 Linux 版本的行为不一致，可靠性差
- 只支持
O_DIRECT模式（绕过页缓存），无法利用内核的缓存优化 - 性能并不理想，在某些场景下甚至不如同步 IO
- 接口复杂，难以使用
1.2 io_uring 的核心设计
io_uring 采用了全新的设计理念，其核心是两个无锁环形队列：
- 提交队列（Submission Queue, SQ）：用户态向内核提交 IO 请求
- 完成队列（Completion Queue, CQ）：内核向用户态返回 IO 完成事件
这两个队列通过 mmap 映射到用户态和内核态的共享内存中，实现了真正的零拷贝。用户程序只需要在共享内存中填写 SQE（Submission Queue Entry），然后通知内核即可；内核完成 IO 后，将 CQE（Completion Queue Entry）写入共享内存，用户程序直接读取结果。整个过程避免了数据在用户态和内核态之间的来回拷贝。
更进一步，io_uring 还支持两种高级模式：
SQ 轮询模式（SQPOLL）：内核创建一个专门的轮询线程，不断检查 SQ 中是否有新的 IO 请求。用户程序提交请求后甚至不需要执行系统调用来通知内核，内核会自动发现并处理。这将系统调用的开销降到了接近于零。
IO 轮询模式（IOPOLL）：对于支持轮询的高性能设备（如 NVMe SSD），内核可以通过轮询设备寄存器来获取 IO 完成状态，而不是等待中断。这进一步降低了延迟，在某些场景下可以将 IO 延迟从几十微秒降到几微秒。
二、项目功能
2.1 项目包含的核心功能
第一层：AsyncIOUring - io_uring 底层封装
- 初始化和管理 io_uring 实例（支持 SQPOLL 和 IOPOLL 模式）
- 提交异步读、写、同步（fsync）等操作
- 独立的完成处理线程，自动回调用户函数
- 独立的提交线程，实现批量提交优化（减少系统调用）
- 完善的 IO 统计信息（请求数、字节数、延迟等）
- 线程安全的请求管理和状态跟踪
第二层：AsyncFileIO - 高层文件 IO 接口
- 自动管理文件描述符的打开和关闭
- 支持多种文件打开模式（只读、只写、读写、追加）
- 提供异步操作（回调风格）和同步操作（future 风格）两种接口
- 支持批量 IO 操作（一次提交多个读写请求）
- 内置 Direct IO 和顺序读取优化
- 自动维护文件偏移量
第三层：AsyncLogger - 高频日志系统
- 内部缓冲区，批量写入减少 IO 次数
- 自动日志文件轮转（按大小）
- 支持多级日志（DEBUG/INFO/WARNING/ERROR）
- 定期刷新线程，确保日志不丢失
- 完全线程安全，支持多线程高并发写入
2.2 可以学习到的核心知识
异步 IO 编程模型
- 理解异步操作的本质：提交 - 回调模式
- 掌握如何设计异步 API（回调风格 vs future 风格 vs 协程风格）
- 学会处理异步操作中的错误传播和异常安全
- 理解异步 IO 中的资源生命周期管理（缓冲区何时可以释放）
io_uring 核心技术
- io_uring 的初始化、配置和清理流程
- SQE 和 CQE 的结构和使用方法
- 批量提交优化：何时提交、提交多少
- 完成事件的处理：轮询 vs 阻塞等待
- SQPOLL 和 IOPOLL 模式的适用场景
高性能编程技巧
- 零拷贝技术：理解
mmap共享内存，避免数据拷贝 - 批量操作：通过减少系统调用次数来提升性能
- 预读优化：使用
posix_fadvise提示内核进行顺序读取优化 - Direct IO：何时绕过页缓存（大文件顺序 IO）
- 队列深度调优：队列深度对性能的影响
并发编程实战
- 线程安全的数据结构：使用
std::atomic和std::mutex保护共享状态 - 生产者 - 消费者模型：提交线程生产请求，完成线程消费结果
- 条件变量的正确使用：避免 busy-waiting，高效等待异步操作完成
- 锁粒度优化：减小临界区范围，提高并发度
系统级编程
- Linux 文件 IO 的各种标志（
O_DIRECT、O_APPEND、O_SYNC等） - 文件描述符的管理和 RAII 封装
fsync与数据持久性保证- 时间戳、性能统计、监控系统的实现
三、核心组件
3.1 AsyncIOUring：io_uring 的完整封装
位于 src/AsyncIOUring.cpp 和 include/AsyncIOUring.h。让我们逐步剖析其设计。
数据结构设计
class AsyncIOUring {
private:
struct io_uring ring_; // io_uring 实例
unsigned queue_depth_; // 队列深度
bool use_sqpoll_; // 是否使用 SQ 轮询
bool use_iopoll_; // 是否使用 IO 轮询
bool initialized_; // 是否已初始化
std::atomic<bool> running_; // 运行标志
std::thread completion_thread_; // 完成处理线程
std::thread submit_thread_; // 提交线程（批量提交优化）
mutable std::mutex request_mutex_; // 请求映射锁
std::unordered_map<uint64_t, std::unique_ptr<IORequest>> pending_requests_;
std::atomic<uint64_t> next_req_id_; // 下一个请求 ID
IOStatsInternal stats_; // IO 统计信息
std::mutex submit_mutex_; // 提交操作互斥锁
std::condition_variable completion_cv_; // 完成条件变量
std::atomic<int> pending_submissions_{0}; // 挂起的提交数量
static constexpr int BATCH_SUBMIT_SIZE = 32; // 批量提交阈值
};
这个设计体现了几个关键思想：
1. 线程分离架构：使用两个独立的后台线程
completion_thread_：专门负责从 CQ 中获取完成事件并执行回调submit_thread_：定期批量提交 SQ 中的请求
这样做的好处是用户线程调用 submitRead/submitWrite 时只需要准备 SQE，立即返回，不会被 IO 阻塞。实际的提交和完成处理都在后台异步进行。
2. 请求跟踪机制：每个 IO 请求都有唯一的 req_id
std::unordered_map<uint64_t, std::unique_ptr<IORequest>> pending_requests_;
当用户提交请求时，系统生成一个 ID 并保存请求信息（包括回调函数）；当 IO 完成时，通过 ID 找回请求并执行回调。这是异步编程的典型模式。
3. 批量提交优化：不是每次调用都立即提交给内核
int pending = pending_submissions_.fetch_add(1, std::memory_order_relaxed) + 1;
// 达到批量大小阈值时立即提交
if (pending >= BATCH_SUBMIT_SIZE) {
int ret = io_uring_submit(&ring_);
pending_submissions_.store(0, std::memory_order_relaxed);
}
这段代码位于 submitRequest 函数中。只有当累积了 32 个请求（BATCH_SUBMIT_SIZE）时，才调用 io_uring_submit 向内核提交。这大大减少了系统调用的次数。同时，submit_thread_ 会每隔 50 微秒检查一次，如果有挂起的请求就提交，避免请求被无限期延迟。
初始化流程
bool AsyncIOUring::initialize() {
if (initialized_) return true;
// 设置 io_uring 参数
unsigned flags = 0;
// SQPOLL 模式：内核线程轮询提交队列，减少用户态到内核态的切换
if (use_sqpoll_) flags |= IORING_SETUP_SQPOLL;
// IOPOLL 模式：轮询 IO 设备完成状态，适合高性能 NVMe SSD
if (use_iopoll_) flags |= IORING_SETUP_IOPOLL;
// 初始化 io_uring 实例
int ret = io_uring_queue_init(queue_depth_, &ring_, flags);
if (ret < 0) {
std::cerr << "io_uring 初始化失败：" << strerror(-ret) << std::endl;
return false;
}
initialized_ = true;
running_ = true;
// 启动完成处理线程
completion_thread_ = std::thread(&AsyncIOUring::completionThread, this);
// 启动提交线程（批量提交优化）
submit_thread_ = std::thread(&AsyncIOUring::submitThread, this);
return true;
}
io_uring_queue_init 是 liburing 库提供的初始化函数，它会：
- 在内核中创建 io_uring 实例
- 分配 SQ 和 CQ 所需的内存
- 将这些内存
mmap映射到用户态 - 设置必要的控制结构
queue_depth 参数决定了可以同时挂起的 IO 操作数量。对于小型应用，128 就足够；对于数据库等高并发场景，建议设置为 512 或 1024。
提交 IO 请求的核心流程
uint64_t AsyncIOUring::submitRead(int fd, void* buffer, size_t length, off_t offset, std::function<void(ssize_t, int)> callback) {
auto request = std::make_unique<IORequest>();
request->op_type = IOOpType::READ;
request->fd = fd;
request->buffer = buffer;
request->length = length;
request->offset = offset;
request->callback = std::move(callback);
request->submit_time = std::chrono::steady_clock::now();
return submitRequest(std::move(request));
}
这里创建了一个 IORequest 对象，记录了所有必要的信息。特别注意 submit_time 字段，它用于后续计算 IO 延迟。
uint64_t AsyncIOUring::submitRequest(std::unique_ptr<IORequest> request) {
if (!initialized_ || !running_) return 0;
// 生成请求 ID
request->req_id = generateRequestId();
std::lock_guard<std::mutex> lock(submit_mutex_);
// 获取提交队列条目（SQE - Submission Queue Entry）
struct io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
if (!sqe) {
std::cerr << "获取 SQE 失败：队列已满" << std::endl;
return 0;
}
// 根据操作类型准备 SQE
switch (request->op_type) {
case IOOpType::READ:
io_uring_prep_read(sqe, request->fd, request->buffer, request->length, request->offset);
break;
case IOOpType::WRITE:
io_uring_prep_write(sqe, request->fd, request->buffer, request->length, request->offset);
break;
case IOOpType::FSYNC:
io_uring_prep_fsync(sqe, request->fd, 0);
break;
default:
return 0;
}
// 将请求 ID 设置为用户数据，用于在完成时识别请求
io_uring_sqe_set_data(sqe, reinterpret_cast<void*>(request->req_id));
// 保存请求 ID
uint64_t req_id = request->req_id;
// 将请求存储到映射中
{
std::lock_guard<std::mutex> ;
pending_requests_[req_id] = std::(request);
}
stats_.total_requests++;
pending = pending_submissions_.(, std::memory_order_relaxed) + ;
(pending >= BATCH_SUBMIT_SIZE) {
ret = (&ring_);
(ret < ) {
std::cerr << << (-ret) << std::endl;
}
pending_submissions_.(, std::memory_order_relaxed);
}
req_id;
}
这段代码的关键步骤：
- 获取 SQE：
io_uring_get_sqe从 SQ 中获取一个空闲的条目。如果队列已满，返回 NULL。 - 准备 SQE：根据操作类型调用相应的
io_uring_prep_*函数。这些函数实际上只是填充 SQE 结构体的字段。 - 设置用户数据：
io_uring_sqe_set_data将请求 ID 作为用户数据附加到 SQE 上。当 IO 完成时，CQE 会携带这个用户数据，从而可以找到对应的请求。 - 保存请求：将请求对象保存到
pending_requests_映射中，以便完成时能找到回调函数。 - 批量提交：只有累积足够多的请求才提交，这是性能优化的关键。
完成处理线程的实现
void AsyncIOUring::completionThread() {
struct io_uring_cqe* cqe;
struct __kernel_timespec timeout;
timeout.tv_sec = 0;
timeout.tv_nsec = 1000000; // 1ms timeout
while (running_) {
// 使用带超时的等待，允许线程定期检查 running_ 标志
int ret = io_uring_wait_cqe_timeout(&ring_, &cqe, &timeout);
if (ret < 0) {
if (ret == -EINTR || ret == -ETIME) {
// 被信号中断或超时，继续循环检查 running_
continue;
}
if (running_) {
std::cerr << "等待完成事件失败：" << strerror(-ret) << std::endl;
}
break;
}
// 处理第一个已经 wait 到的 CQE
handleCompletion(cqe);
io_uring_cqe_seen(&ring_, cqe);
// 批量处理所有其他可用的完成事件（不阻塞）
unsigned head;
unsigned count = 0;
io_uring_for_each_cqe(&ring_, head, cqe) {
handleCompletion(cqe);
count++;
// 限制单次批量处理数量，避免长时间持有 CPU
if (count >= 128) break;
}
if (count > 0) {
// 批量标记所有 CQE 已处理（减少系统调用）
io_uring_cq_advance(&ring_, count);
}
// 所有 CQE 处理完后，只通知一次（避免频繁唤醒）
completion_cv_.();
}
}
1. 超时等待：使用 1ms 超时而不是无限等待。这样可以在析构时能够及时退出线程，避免死锁。
2. 批量处理：不是处理一个 CQE 就立即返回，而是尝试一次性处理所有可用的 CQE。io_uring_for_each_cqe 是一个宏，它遍历 CQ 中所有已就绪的条目。这样做的好处是：
- 减少循环次数
- 减少系统调用（
io_uring_cq_advance一次性标记多个 CQE 为已读） - 提高吞吐量
3. 限制批量大小：最多一次处理 128 个 CQE。这是为了避免在极高负载下，完成线程长时间占用 CPU，影响其他线程的运行。
void AsyncIOUring::handleCompletion(struct io_uring_cqe* cqe) {
// 从用户数据中获取请求 ID
uint64_t req_id = reinterpret_cast<uint64_t>(io_uring_cqe_get_data(cqe));
if (req_id == 0) return; // 无效请求
std::unique_ptr<IORequest> request;
// 优化锁粒度，减少临界区时间
{
std::lock_guard<std::mutex> lock(request_mutex_);
auto it = pending_requests_.find(req_id);
if (it == pending_requests_.end()) return; // 请求已被处理或不存在
request = std::move(it->second);
pending_requests_.erase(it);
}
// 锁在这里释放，后续操作不需要持有锁
// 获取 IO 操作结果
ssize_t result = cqe->res; // 正数表示成功字节数，负数表示错误码
int error = (result < 0) ? -result : 0;
// 快速路径：只更新计数器，不计算延迟（减少开销）
uint64_t completed = stats_.completed_requests.fetch_add(1, std::memory_order_relaxed) + 1;
if (error != 0) {
stats_.failed_requests.fetch_add(1, std::memory_order_relaxed);
} else {
if (request->op_type == IOOpType::READ) {
stats_.total_bytes_read.fetch_add(result, std::memory_order_relaxed);
} (request->op_type == IOOpType::WRITE) {
stats_.total_bytes_written.(result, std::memory_order_relaxed);
}
}
((completed & ) == ) {
now = std::chrono::steady_clock::();
latency = std::chrono::<std::chrono::microseconds>(now - request->submit_time).();
old_latency = stats_.avg_latency_us.(std::memory_order_relaxed);
new_latency = (old_latency * + latency) / ;
stats_.avg_latency_us.(new_latency, std::memory_order_relaxed);
}
(request->callback) {
{
request->(result, error);
} ( std::exception& e) {
std::cerr << << e.() << std::endl;
}
}
}
这段代码展示了几个重要的性能优化技巧：
1. 锁粒度优化：只在查找和删除请求时持有锁，然后立即释放。回调函数的执行是在锁外进行的，这样可以：
- 避免回调函数中的耗时操作阻塞其他完成事件的处理
- 避免回调函数中调用
submitRead/submitWrite导致的死锁（因为它们也需要获取锁）
2. 延迟采样统计：计算每个请求的延迟需要调用 std::chrono::steady_clock::now()，这个操作在某些 CPU 上可能需要几十纳秒。为了减少开销，代码只在每 64 个请求时采样一次。使用位运算 (completed & 0x3F) == 0 来判断是否是 64 的倍数，这比 completed % 64 == 0 快得多。
3. 指数移动平均：使用简单的 (old * 9 + new) / 10 公式来计算平均延迟。这比维护一个延迟历史队列要高效得多，同时对最近的数据赋予更高的权重。
3.2 AsyncFileIO：易用的高层封装
AsyncFileIO 类在 AsyncIOUring 的基础上，提供了更符合日常使用习惯的文件操作接口。
文件句柄管理
class AsyncFileHandle {
public:
AsyncFileHandle(int fd, const std::string& path)
: fd_(fd), path_(path), offset_(0) {}
~AsyncFileHandle() {
if (fd_ >= 0) close(fd_);
}
int getFd() const { return fd_; }
const std::string& getPath() const { return path_; }
off_t getOffset() const { return offset_; }
void setOffset(off_t offset) { offset_ = offset; }
private:
int fd_;
std::string path_;
off_t offset_;
};
这个类使用 RAII（Resource Acquisition Is Initialization）模式，确保文件描述符在对象销毁时自动关闭。同时维护了当前的文件偏移量，这样用户不需要手动跟踪读写位置。
文件打开与优化
std::shared_ptr<AsyncFileHandle> AsyncFileIO::openFile(const std::string& path, FileOpenMode mode) {
std::lock_guard<std::mutex> lock(file_mutex_);
int flags = modeToFlags(mode);
// Direct IO 标志（绕过页缓存）
if (use_direct_io_) flags |= O_DIRECT;
// 打开文件
int fd = open(path.c_str(), flags, 0644);
if (fd < 0) {
std::cerr << "打开文件失败 [" << path << "]: " << strerror(errno) << std::endl;
return nullptr;
}
// 优化顺序读取性能
// 对于只读和读写模式，启用顺序访问提示，让内核执行激进的预读
if (!use_direct_io_ && (mode == FileOpenMode::READ_ONLY || mode == FileOpenMode::READ_WRITE)) {
// POSIX_FADV_SEQUENTIAL：告诉内核这是顺序访问，应该积极预读并快速释放已读页面
posix_fadvise(fd, 0, 0, POSIX_FADV_SEQUENTIAL);
}
return std::make_shared<AsyncFileHandle>(fd, path);
}
这里的 posix_fadvise 调用非常重要。POSIX_FADV_SEQUENTIAL 标志告诉内核：
- 这个文件会被顺序读取
- 应该激进地进行预读（read-ahead），提前将后续数据加载到页缓存
- 已经读过的页面可以快速释放，因为不太可能再次访问
异步读写的包装
uint64_t AsyncFileIO::readAsync(std::shared_ptr<AsyncFileHandle> handle, void* buffer, size_t length, std::function<void(ssize_t, int)> callback) {
if (!handle) return 0;
off_t offset = handle->getOffset();
// 提交读取请求，并在完成后更新偏移量
auto wrapped_callback = [handle, offset, length, callback](ssize_t result, int error) {
if (result > 0) handle->setOffset(offset + result);
if (callback) callback(result, error);
};
return io_uring_->submitRead(handle->getFd(), buffer, length, offset, wrapped_callback);
}
这里使用了回调包装（callback wrapping）技巧：
- 捕获当前的文件偏移量
- 创建一个新的回调函数，它会在 IO 完成后自动更新偏移量
- 然后调用用户的回调函数
同步接口的实现
std::future<ssize_t> AsyncFileIO::readSync(std::shared_ptr<AsyncFileHandle> handle, void* buffer, size_t length) {
// 创建 promise 和 future 对
auto promise = std::make_shared<std::promise<ssize_t>>();
std::future<ssize_t> future = promise->get_future();
// 提交异步读取，使用 promise 设置结果
readAsync(handle, buffer, length, [promise](ssize_t result, int error) {
if (error != 0) promise->set_value(-error);
else promise->set_value(result);
});
return future;
}
这是一个经典的异步到同步的转换模式：
- 创建一个
std::promise和对应的std::future - 在异步回调中，将结果通过
promise->set_value()传递 - 返回
future给用户，用户可以调用future.get()来阻塞等待结果
批量 IO 操作
void AsyncFileIO::readBatch(std::shared_ptr<AsyncFileHandle> handle,
const std::vector<void*>& buffers,
const std::vector<size_t>& lengths,
std::function<void(std::vector<ssize_t>)> callback) {
if (buffers.size() != lengths.size()) return;
// 使用 shared_ptr 管理结果向量，确保在所有回调完成前不被销毁
auto results = std::make_shared<std::vector<ssize_t>>(buffers.size(), 0);
auto counter = std::make_shared<std::atomic<size_t>>(buffers.size());
for (size_t i = 0; i < buffers.size(); ++i) {
readAsync(handle, buffers[i], lengths[i], [results, counter, callback, i](ssize_t result, int error) {
(*results)[i] = (error != 0) ? -error : result;
// 当所有操作完成时，调用用户回调
if (counter->fetch_sub(1) == 1) {
if (callback) callback(*results);
}
});
}
}
这个函数展示了如何在异步环境中实现批量操作：
- 创建一个共享的结果向量和计数器
- 为每个子操作提交异步请求
- 在每个子操作的回调中，保存结果并递减计数器
- 当计数器归零时（所有子操作都完成），调用用户的回调函数
注意这里使用 std::shared_ptr 来管理 results 和 counter，这样可以确保即使某些回调延迟执行，这些对象也不会被提前销毁。
3.3 AsyncLogger：生产级异步日志系统
AsyncLogger 是本项目的实战应用示例，展示了如何使用异步 IO 构建一个高性能的日志系统。
核心数据结构
class AsyncLogger {
private:
std::unique_ptr<AsyncFileIO> file_io_;
std::shared_ptr<AsyncFileHandle> current_file_;
std::string log_dir_;
size_t max_file_size_;
std::atomic<size_t> current_file_size_; // 使用原子变量保证线程安全
std::vector<char> buffer_;
size_t buffer_pos_;
std::mutex buffer_mutex_;
std::atomic<uint64_t> total_logs_;
std::atomic<uint64_t> total_bytes_;
std::thread flush_thread_;
std::atomic<bool> running_;
};
设计要点：
- 内部缓冲区：
buffer_用于暂存日志，避免每条日志都触发一次 IO - 原子计数器：使用
std::atomic来统计日志数量和字节数，避免锁竞争 - 定期刷新线程：
flush_thread_每隔 1 秒自动将缓冲区刷新到磁盘
日志写入流程
void AsyncLogger::log(LogLevel level, const std::string& message) {
// 构造日志行：[时间戳] [级别] 消息\n
std::stringstream ss;
ss << "[" << getCurrentTimestamp() << "] "
<< "[" << getLevelString(level) << "] " << message << "\n";
std::string log_line = ss.str();
std::lock_guard<std::mutex> lock(buffer_mutex_);
// 如果缓冲区空间不足，先刷新
if (buffer_pos_ + log_line.size() > buffer_.size()) flushBuffer();
memcpy(buffer_.data() + buffer_pos_, log_line.c_str(), log_line.size());
buffer_pos_ += log_line.size();
total_logs_++;
}
这里的关键设计是先缓冲，再批量写入。每条日志只是追加到内存缓冲区，不触发 IO。只有当：
- 缓冲区满了
- 或者定期刷新线程触发
才会真正写入磁盘。这大大减少了 IO 次数，提高了性能。
void AsyncLogger::flushBuffer() {
// 注意：此函数必须在持有 buffer_mutex_ 的情况下调用
if (!current_file_ || buffer_pos_ == 0) return;
// 检查是否需要轮转日志文件
if (current_file_size_.load(std::memory_order_relaxed) + buffer_pos_ > max_file_size_) {
rotateLogFile();
}
// 使用 shared_ptr 避免大量内存拷贝
auto data_copy = std::make_shared<std::vector<char>>(buffer_.begin(), buffer_.begin() + buffer_pos_);
size_t size = buffer_pos_;
file_io_->writeAsync(current_file_, data_copy->data(), size, [this, size, data_copy](ssize_t result, int error) {
if (error == 0 && result > 0) {
// 使用原子操作更新文件大小
current_file_size_.fetch_add(result, std::memory_order_relaxed);
total_bytes_.fetch_add(result, std::memory_order_relaxed);
}
});
buffer_pos_ = 0;
}
这里有两个重要的技巧：
1. 数据拷贝：由于 writeAsync 是异步的，它可能在 flushBuffer 返回后才真正执行。因此不能直接传递 buffer_ 的指针，必须拷贝一份数据。这里使用 std::shared_ptr<std::vector<char>>，并在回调中捕获它，确保数据在 IO 完成前不会被销毁。
2. 原子操作：current_file_size_ 使用 std::atomic，因为它会在异步回调中被更新，而回调可能在不同的线程执行。
日志文件轮转
void AsyncLogger::rotateLogFile() {
// 生成新的日志文件名：log_YYYYMMDD_HHMMSS.log
auto now = std::time(nullptr);
auto tm = std::localtime(&now);
std::stringstream ss;
ss << log_dir_ << "/log_" << std::put_time(tm, "%Y%m%d_%H%M%S") << ".log";
std::string new_log_path = ss.str();
// 关闭旧文件（如果有）
if (current_file_) {
file_io_->flushSync(current_file_).get();
current_file_.reset();
}
// 打开新文件
current_file_ = file_io_->openFile(new_log_path, FileOpenMode::APPEND);
current_file_size_.store(0, std::memory_order_relaxed);
std::cout << "日志轮转：" << new_log_path << std::endl;
}
当日志文件达到设定的最大大小（默认 100MB）时，会自动创建一个新文件。文件名包含时间戳，便于归档和查找。
四、示例程序深度解读
4.1 database_sim.cpp：模拟数据库存储引擎
这个示例展示了如何使用异步 IO 构建一个简单的键值存储引擎，类似于 RocksDB 或 LevelDB 的简化版本。
记录结构设计
// 定义记录大小（固定大小简化实现）
constexpr size_t KEY_SIZE = 32;
constexpr size_t VALUE_SIZE = 256;
constexpr size_t RECORD_SIZE = KEY_SIZE + VALUE_SIZE;
struct Record {
char key[KEY_SIZE];
char value[VALUE_SIZE];
Record() {
memset(key, 0, KEY_SIZE);
memset(value, 0, VALUE_SIZE);
}
Record(const std::string& k, const std::string& v) {
memset(key, 0, KEY_SIZE);
memset(value, 0, VALUE_SIZE);
strncpy(key, k.c_str(), KEY_SIZE - 1);
strncpy(value, v.c_str(), VALUE_SIZE - 1);
}
};
采用固定大小的记录格式，这样可以通过简单的计算直接定位任意记录的文件偏移量：offset = record_index * RECORD_SIZE。这是许多数据库索引结构（如 B+ 树）的基础。
索引管理
class SimpleKVStore {
private:
std::unordered_map<std::string, off_t> index_; // key -> 文件偏移量
size_t record_count_;
void loadIndex() {
// 获取文件大小
struct stat st;
if (stat(db_path_.c_str(), &st) != 0) return;
size_t file_size = st.st_size;
size_t num_records = file_size / RECORD_SIZE;
// 同步读取所有记录，构建索引
for (size_t i = 0; i < num_records; ++i) {
Record record;
data_file_->setOffset(i * RECORD_SIZE);
auto future = file_io_->readSync(data_file_, &record, RECORD_SIZE);
ssize_t result = future.get();
if (result > 0) {
std::string key = record.getKey();
if (!key.empty()) {
index_[key] = i * RECORD_SIZE;
}
}
}
record_count_ = num_records;
}
};
这是一个简化的索引实现。在初始化时，读取整个文件，将每个记录的 key 和其文件偏移量建立映射。真实的数据库会使用更复杂的结构（如 B+ 树）来避免加载整个索引到内存。
批量写入
void SimpleKVStore::batchWrite(const std::vector<Record>& records, std::function<void(bool)> callback) {
auto start_time = std::chrono::high_resolution_clock::now();
// 准备批量写入的缓冲区
std::vector<const void*> buffers;
std::vector<size_t> lengths;
// 记录起始偏移量
off_t start_offset = record_count_ * RECORD_SIZE;
for (const auto& record : records) {
// 需要复制记录到堆上，因为异步 IO 可能在函数返回后才完成
auto* record_copy = new Record(record);
buffers.push_back(record_copy);
lengths.push_back(RECORD_SIZE);
// 更新索引
std::string key = record.getKey();
off_t offset = start_offset + (buffers.size() - 1) * RECORD_SIZE;
index_[key] = offset;
}
// 更新记录数
record_count_ += records.size();
// 批量异步写入
file_io_->writeBatch(data_file_, buffers, lengths, [this, buffers, start_time, callback](std::vector<ssize_t> results) {
// 清理缓冲区
for (auto* buffer : buffers) delete static_cast<const Record*>(buffer);
// 检查结果
bool success = true;
size_t total_written = ;
( result : results) {
(result < ) { success = ; ; }
total_written += result;
}
end_time = std::chrono::high_resolution_clock::();
duration = std::chrono::<std::chrono::microseconds>(end_time - start_time).();
(success) {
std::cout << << results.() << << total_written << << duration << << std::endl;
}
(callback) (success);
});
}
批量写入的优势：
- 一次性提交多个 IO 请求，充分利用磁盘的并行处理能力
- 减少系统调用次数
- 在支持 NCQ（Native Command Queuing）的 SSD 上，可以显著提升性能
测试场景
void testMixedWorkload(SimpleKVStore& db, int duration_seconds) {
std::cout << "\n========== 混合读写测试 ==========" << std::endl;
std::cout << "运行时长：" << duration_seconds << " 秒" << std::endl;
std::cout << "工作负载：70% 读取，30% 写入" << std::endl;
std::atomic<bool> running{true};
std::atomic<uint64_t> total_reads{0};
std::atomic<uint64_t> total_writes{0};
// 启动多个工作线程
std::vector<std::thread> workers;
int num_workers = 4;
for (int i = 0; i < num_workers; ++i) {
workers.emplace_back([&, i]() {
std::random_device rd;
std::mt19937 gen(rd());
std::uniform_int_distribution<> op_dis(0, 99);
std::uniform_int_distribution<> id_dis(0, 10000);
while (running) {
int op = op_dis(gen);
if (op < 70) { // 70% 读取操作
std::string key = "key_" + std::to_string(id_dis(gen));
db.read(key);
total_reads++;
} else { // 30% 写入操作
std::vector<Record> records;
records.push_back(generateRandomRecord((gen)));
db.(records, []( success) {});
total_writes++;
}
std::this_thread::(std::chrono::());
}
});
}
;
std::this_thread::(std::chrono::(duration_seconds));
running = ;
(& worker : workers) worker.();
monitor.();
}
这个测试模拟了真实数据库的工作负载：
- 70% 的操作是读取（模拟查询）
- 30% 的操作是写入（模拟插入/更新）
- 4 个工作线程并发访问
- 实时监控每秒的操作数
4.2 high_freq_logger.cpp：对比同步/异步性能
这个示例直观地展示了异步 IO 的性能优势。
同步日志基准
void synchronousLoggingTest(int num_logs, int num_threads) {
std::cout << "\n========== 同步 IO 日志写入测试 ==========" << std::endl;
auto start_time = std::chrono::high_resolution_clock::now();
// 打开日志文件
FILE* log_file = fopen("./logs/sync_test.log", "w");
// 多线程生成日志并同步写入
std::vector<std::thread> threads;
std::mutex file_mutex;
for (int t = 0; t < num_threads; ++t) {
threads.emplace_back([&, t]() {
int logs_per_thread = num_logs / num_threads;
for (int i = 0; i < logs_per_thread; ++i) {
auto log = generateRandomLog();
std::string log_str = formatLog(log) + "\n";
// 同步写入（需要加锁）
std::lock_guard<std::mutex> lock(file_mutex);
fwrite(log_str.c_str(), 1, log_str.size(), log_file);
}
});
}
for (auto& thread : threads) thread.join();
fflush(log_file);
fclose(log_file);
auto end_time = std::chrono::high_resolution_clock::now();
auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time).count();
std::cout << "完成时间：" << duration << << std::endl;
std::cout << << (num_logs * / duration) << << std::endl;
}
同步方案的问题：
- 每次
fwrite都可能触发一次系统调用 - 多线程需要用锁保护文件指针，导致线程间竞争
- 业务线程被 IO 阻塞，无法继续处理请求
异步日志实现
void asynchronousLoggingTest(int num_logs, int num_threads) {
std::cout << "\n========== 异步 IO 日志写入测试 ==========" << std::endl;
auto start_time = std::chrono::high_resolution_clock::now();
// 创建异步日志器
AsyncLogger logger("./logs", 50 * 1024 * 1024, 64 * 1024); // 64KB 缓冲区
if (!logger.initialize()) {
std::cerr << "初始化异步日志器失败" << std::endl;
return;
}
// 多线程生成日志并异步写入
std::vector<std::thread> threads;
std::atomic<int> completed_logs{0};
for (int t = 0; t < num_threads; ++t) {
threads.emplace_back([&, t]() {
int logs_per_thread = num_logs / num_threads;
for (int i = 0; i < logs_per_thread; ++i) {
auto log = generateRandomLog();
std::string log_str = formatLog(log);
// 异步写入（无需加锁，内部已处理并发）
logger.log(AsyncLogger::LogLevel::INFO, log_str);
completed_logs++;
}
});
}
for (auto& thread : threads) thread.join();
// 确保所有日志写入完成
logger.flush();
auto end_time = std::chrono::high_resolution_clock::();
duration = std::chrono::<std::chrono::milliseconds>(end_time - start_time).();
std::cout << << duration << << std::endl;
std::cout << << (num_logs * / duration) << << std::endl;
}
异步方案的优势：
- 业务线程只需将日志追加到内存缓冲区，立即返回（几十纳秒）
- 实际的 IO 在后台线程批量执行，减少系统调用
- 无锁设计（只在操作缓冲区时需要锁）
在实测中，异步方案的吞吐量是同步方案的3-5 倍。
总结
io_uring 作为 Linux 内核的最新异步 IO 接口，代表了未来的方向。它不仅支持文件 IO，还支持网络 IO、定时器、信号等几乎所有的 IO 操作。无论你是想为自己的项目添加高性能 IO 能力，还是希望深入学习系统编程和性能优化，这都是一个非常适合的项目！
