深入了解Redis线程工作模式（redis线程工作模式）
深入了解Redis线程工作模式
Redis作为一种高性能的key-value存储系统，无疑成为了热门技术之一。Redis快速处理大量数据的能力取决于其高效的线程工作模式。本文将深入了解Redis的线程工作模式，以便更好的理解Redis的内部工作原理。
Redis的线程模型
Redis从2.8版本开始，采用了多线程方式实现I/O多路复用，用于处理并发连接请求。Redis线程数默认为cpu核心数量。其中，主线程被用于处理客户端的连接和协议解析，以及同步的I/O操作。工作线程负责处理异步的 I/O 操作（如文件事件），并执行在key空间中定义的命令。下图是Redis多线程架构示意图：
如上图所示，Redis主线程接收客户端请求，同时与工作线程协作处理I/O读写。工作线程也处理Redis持久化操作和其他底层任务。Redis线程模型的优点是减少锁竞争，在高并发下性能得到强力保障。
Redis的I/O模型
Redis使用了epoll进行I/O多路复用，以实现高并发读写。epoll采取了边缘触发(EPOLLET)的方式，只有当文件描述符状态发生变化时，内核才会通知应用程序。
Redis采用epoll的ET模式以减少CPU上下文切换，处理过程简单，效率高，并且避免了遗漏的文件描述符事件。ET模式需要确保读取或写入的缓存区，都应该被完整的处理一遍。因此，当Redis请求的大小超过系统默认的缓冲区大小时，redis在异步读取/写入时，需要进行多次的读写操作，以确保缓冲区文本被完整地读出/写入。以下是Redis ET读写实现过程的示意图：
Redis的AOF持久化模式
Redis主要支持两种持久化方式：RDB和AOF。不过本文只讨论AOF持久化策略，它也是Redis使用最多的一种策略。AOF的写操作可能会卡住主进程，从而影响性能。为了降低写操作的影响，Redis容器采取了异步AOF持久化模式，即BGREWRITEAOF（后台重写）。该模式通过生成新的AOF文件，然后将其追加到原AOF文件中。该过程在后台完成，不会阻塞Redis的主工作线程。以下为Redis AOF持久化模式示意图：
从上图中可以看出，Redis将不太重要的任务交给了工作线程，并将I/O操作异步化，以避免重要任务的阻塞。同时，Redis采取了异步AOF持久化模式，以减轻写操作所产生的性能影响。Redis的线程工作模式十分高效，可接受高并发请求，具有出色的稳定性和高可用性。
本文示例代码：
// 1. mn thread
aeMn()
{
// 等待事件
aeApiPoll();
// …
// 接收到 read 命令后
if (socket_read_pending()) {
// 将命令交给工作线程执行
worker_pool.schedule(read_handler, socket_fd);
}
// …
// 在工作线程中处理异步 I/O 请求
worker_pool.run_io_threads();
// …
}
// 2. 工作线程
void run_io_threads()
{
// 利用 poll() 函数进入等待就绪的状态
// 和 poll() 函数类似，这里我们用自己封装的 async_poll() 函数
while(true){
// 处理正在执行的 IO 请求
process_io_requests();
// 查询当前是否有需要执行的操作
if (async_poll(thread_read_fds)) {
// 执行每个客户端的命令
process_read_commands();
// 执行文件事件
process_disk_events();
}
// 执行 Redis 的一些常规操作
process_checkpointing();
process_timers();
process_scroll();
}
}
// 3. 异步 AOF 备份操作
void bg_rewrite_aof()
{
// 创建新文件
aof_file = open(“redis.aof.new”, O_CREAT | O_RDWR, S_IRWXU);
// …
// 异步写入
while(true) {
append_to_file(io_buffer, io_size);
// 写入速度过快
if (a_socket_send_buffer_full()) {
// 等待套接字写空
a_socket_send_wt_until_unblock();
}
// 压缩并移动文件
compress_and_transfer_approved_files();
}
}
// 4. Redis 的多线程模块
class WorkerPool {
public:
// 从工作队列中取出 worker 开始执行
void run_worker(Worker& worker) {
pthread_create(worker.tid, NULL, worker.worker_routine, (void*)&tpool);
}
};
int mn() {
WorkerPool tp;
Worker* threads[THREAD_NUM];
for (int i = 0; i
threads[i].tid = i;
threads[i].worker_routine = worker_routine;
tp.run_worker(threads[i]);
}
}
