redis7.0源码阅读(四):Redis中的IO多线程(线程池)

本文涉及的产品
云数据库 Redis 版,社区版 2GB
推荐场景:
搭建游戏排行榜
简介: redis7.0源码阅读(四):Redis中的IO多线程(线程池)

一、Redis中的IO多线程原理

服务端收到一条信息,给它deconde成一条命令

然后根据命令获得一个结果(reply)

然后将结果encode后,发送回去

redis的单线程是指,命令执行(logic)都是在单线程中运行的

接受数据read和发送数据write都是可以在io多线程(线程池)中去运行

在Redis中,生产者也可以作为消费者,反之亦然,没有明确界限。

二、设置io多线程(调试设置)

在redis.conf中

设置io-threads-do-reads yes就可以开启io多线程

设置io-threads 2,设置为2(为了方便调试,真正使用的时候,可以根据需要设置),其中一个为主线程,另外一个是io线程

在networking.c中找到stopThreadedIOIfNeeded,如果在redis-cli中输入一条命令,是不会执行多线程的,因为它会判断,如果pending(需要做的命令)个数比io线程数少,就不会执行多线程

因此提前return 0,确保执行多线程,便于调试

int stopThreadedIOIfNeeded(void) {
    int pending = listLength(server.clients_pending_write);
    /* Return ASAP if IO threads are disabled (single threaded mode). */
    if (server.io_threads_num == 1) return 1;
    return 0;//为了调试,提前退出(自己添加的一行)
    if (pending < (server.io_threads_num*2)) {
        if (server.io_threads_active) stopThreadedIO();
        return 1;
    } else {
        return 0;
    }
}

到此为止,只需要,运行redis-server,在networking.c的 readQueryFromClient中打个断点,然后在redis-cli中输入任意set key value就可以进入io多线程,进行调试

下图可以看到箭头指向的两个线程,一个是主线程,另一个是io线程

三、Redis中的IO线程池

1、读取任务readQueryFromClient

postponeClientRead(c)就是判断io多线程模式,并将任务添加到 任务队列中

void readQueryFromClient(connection *conn) { 
    client *c = connGetPrivateData(conn);
    int nread, big_arg = 0;
    size_t qblen, readlen;
    /* Check if we want to read from the client later when exiting from
     * the event loop. This is the case if threaded I/O is enabled. */
    if (postponeClientRead(c)) return; 
  //后面省略......
}

2、主线程将 待读客户端 添加到Read任务队列(生产者)postponeClientRead

如果是io多线程模式,那么将任务添加到任务队列。

(这个函数名的意思,延迟读,就是将任务加入到任务队列,后续去执行)

int postponeClientRead(client *c) {
    if (server.io_threads_active &&
        server.io_threads_do_reads &&
        !ProcessingEventsWhileBlocked &&
        !(c->flags & (CLIENT_MASTER|CLIENT_SLAVE|CLIENT_BLOCKED)) &&
        io_threads_op == IO_THREADS_OP_IDLE)
    {
        listAddNodeHead(server.clients_pending_read,c);//往任务队列中插入任务
        c->pending_read_list_node = listFirst(server.clients_pending_read);
        return 1;
    } else {
        return 0;
    }
}

3、多线程Read IO任务 handleClientsWithPendingReadsUsingThreads

基本原理和多线程Write IO是一样的,直接看多线程Write IO就行了。

其中processInputBuffer是解析协议

int handleClientsWithPendingReadsUsingThreads(void) {
    if (!server.io_threads_active || !server.io_threads_do_reads) return 0;
    int processed = listLength(server.clients_pending_read);
    if (processed == 0) return 0;
    /* Distribute the clients across N different lists. */
    listIter li;
    listNode *ln;
    listRewind(server.clients_pending_read,&li);
    int item_id = 0;
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        int target_id = item_id % server.io_threads_num;
        listAddNodeTail(io_threads_list[target_id],c);
        item_id++;
    }
    /* Give the start condition to the waiting threads, by setting the
     * start condition atomic var. */
    io_threads_op = IO_THREADS_OP_READ;
    for (int j = 1; j < server.io_threads_num; j++) {
        int count = listLength(io_threads_list[j]);
        setIOPendingCount(j, count);
    }
    /* Also use the main thread to process a slice of clients. */
    listRewind(io_threads_list[0],&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        readQueryFromClient(c->conn);
    }
    listEmpty(io_threads_list[0]);
    /* Wait for all the other threads to end their work. */
    while(1) {
        unsigned long pending = 0;
        for (int j = 1; j < server.io_threads_num; j++)
            pending += getIOPendingCount(j);
        if (pending == 0) break;
    }
    io_threads_op = IO_THREADS_OP_IDLE;
    /* Run the list of clients again to process the new buffers. */
    while(listLength(server.clients_pending_read)) {
        ln = listFirst(server.clients_pending_read);
        client *c = listNodeValue(ln);
        listDelNode(server.clients_pending_read,ln);
        c->pending_read_list_node = NULL;
        serverAssert(!(c->flags & CLIENT_BLOCKED));
        if (beforeNextClient(c) == C_ERR) {
            /* If the client is no longer valid, we avoid
             * processing the client later. So we just go
             * to the next. */
            continue;
        }
        /* Once io-threads are idle we can update the client in the mem usage buckets */
        updateClientMemUsageBucket(c);
        if (processPendingCommandsAndResetClient(c) == C_ERR) {
            /* If the client is no longer valid, we avoid
             * processing the client later. So we just go
             * to the next. */
            continue;
        }
        if (processInputBuffer(c) == C_ERR) {
            /* If the client is no longer valid, we avoid
             * processing the client later. So we just go
             * to the next. */
            continue;
        }
        /* We may have pending replies if a thread readQueryFromClient() produced
         * replies and did not install a write handler (it can't).
         */
        if (!(c->flags & CLIENT_PENDING_WRITE) && clientHasPendingReplies(c))
            clientInstallWriteHandler(c);
    }
    /* Update processed count on server */
    server.stat_io_reads_processed += processed;
    return processed;
}

4、多线程write IO任务(消费者)handleClientsWithPendingWritesUsingThreads

1.判断是否有必要开启IO多线程

2.如果没启动IO多线程,就启动IO多线程

3.负载均衡:write任务队列,均匀分给不同io线程

4.启动io子线程

5.主线程执行io任务

6.主线程等待io线程写结束

/* This function achieves thread safety using a fan-out -> fan-in paradigm:
 * Fan out: The main thread fans out work to the io-threads which block until
 * setIOPendingCount() is called with a value larger than 0 by the main thread.
 * Fan in: The main thread waits until getIOPendingCount() returns 0. Then
 * it can safely perform post-processing and return to normal synchronous
 * work. */
int handleClientsWithPendingWritesUsingThreads(void) {
    int processed = listLength(server.clients_pending_write);
    if (processed == 0) return 0; /* Return ASAP if there are no clients. */
    /* If I/O threads are disabled or we have few clients to serve, don't
     * use I/O threads, but the boring synchronous code. */
    if (server.io_threads_num == 1 || stopThreadedIOIfNeeded()) {//判断是否有必要开启IO多线程
        return handleClientsWithPendingWrites();
    }
    /* Start threads if needed. */
    if (!server.io_threads_active) startThreadedIO();//开启io多线程
    /* Distribute the clients across N different lists. */
    listIter li;
    listNode *ln;
    listRewind(server.clients_pending_write,&li);//创建一个迭代器li,用于遍历任务队列clients_pending_write
    int item_id = 0;//默认是0,先分配给主线程去做(生产者也可能是消费者),如果设置成1,则先让io线程1去做
    //io_threads_list[0] 主线程
    //io_threads_list[1] io线程
    //io_threads_list[2] io线程   
    //io_threads_list[3] io线程   
    //io_threads_list[4] io线程
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);//取出一个任务
        c->flags &= ~CLIENT_PENDING_WRITE;
        /* Remove clients from the list of pending writes since
         * they are going to be closed ASAP. */
        if (c->flags & CLIENT_CLOSE_ASAP) {//表示该客户端的输出缓冲区超过了服务器允许范围,将在下一次循环进行一个关闭,也不返回任何信息给客户端,删除待读客户端
            listDelNode(server.clients_pending_write, ln);
            continue;
        }
        /* Since all replicas and replication backlog use global replication
         * buffer, to guarantee data accessing thread safe, we must put all
         * replicas client into io_threads_list[0] i.e. main thread handles
         * sending the output buffer of all replicas. */
        if (getClientType(c) == CLIENT_TYPE_SLAVE) {
            listAddNodeTail(io_threads_list[0],c);
            continue;
        }
        //负载均衡:将任务队列中的任务 添加 到不同的线程消费队列中去,每个线程就可以从当前线程的消费队列中取任务就行了
        //这样做的好处是,避免加锁。当前是在主线程中,进行分配任务
        //通过取余操作,将任务均分给不同io线程
        int target_id = item_id % server.io_threads_num;
        listAddNodeTail(io_threads_list[target_id],c);
        item_id++;
    }
    /* Give the start condition to the waiting threads, by setting the
     * start condition atomic var. */
    io_threads_op = IO_THREADS_OP_WRITE;
    for (int j = 1; j < server.io_threads_num; j++) {
        int count = listLength(io_threads_list[j]);
        setIOPendingCount(j, count);//设置io线程启动条件,启动io线程
    }
    /* Also use the main thread to process a slice of clients. */
    listRewind(io_threads_list[0],&li);//让主线程去处理一部分任务(io_threads_list[0])
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        writeToClient(c,0);
    }
    listEmpty(io_threads_list[0]);
    /* Wait for all the other threads to end their work. */
    while(1) {//剩下的任务io_threads_list[1],io_threads_list[2].....给io线程去做,等待io线程完成任务
        unsigned long pending = 0;
        for (int j = 1; j < server.io_threads_num; j++)
            pending += getIOPendingCount(j);//等待io线程结束,并返回处理的数量
        if (pending == 0) break;
    }
    io_threads_op = IO_THREADS_OP_IDLE;
    /* Run the list of clients again to install the write handler where
     * needed. */
    listRewind(server.clients_pending_write,&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        /* Update the client in the mem usage buckets after we're done processing it in the io-threads */
        updateClientMemUsageBucket(c);
        /* Install the write handler if there are pending writes in some
         * of the clients. */
        if (clientHasPendingReplies(c) &&
                connSetWriteHandler(c->conn, sendReplyToClient) == AE_ERR)
        {
            freeClientAsync(c);
        }
    }
    listEmpty(server.clients_pending_write);
    /* Update processed count on server */
    server.stat_io_writes_processed += processed;
    return processed;
}

负载均衡:将任务队列中的任务 添加 到不同的线程消费队列中去,每个线程就可以从当前线程的消费队列中取任务就行了。这样做的好处是,避免加锁。当前是在主线程中,进行分配任务通过取余操作,将任务均分给不同的io线程。

四、线程调度

1、开启io线程startThreadedIO

每个io线程都有一把锁,如果主线程把锁还回去了,那么io线程就会启动,不再阻塞

并设置io线程标识为活跃状态io_threads_active=1

void startThreadedIO(void) {
    serverAssert(server.io_threads_active == 0);
    for (int j = 1; j < server.io_threads_num; j++)
        pthread_mutex_unlock(&io_threads_mutex[j]);
    server.io_threads_active = 1;
}

2、关闭io线程stopThreadedIO

每个io线程都有一把锁,如果主线程拿了,那么io线程就会阻塞等待,也就是停止了IO线程

并设置io线程标识为非活跃状态io_threads_active=0

void stopThreadedIO(void) {
    /* We may have still clients with pending reads when this function
     * is called: handle them before stopping the threads. */
    handleClientsWithPendingReadsUsingThreads();
    serverAssert(server.io_threads_active == 1);
    for (int j = 1; j < server.io_threads_num; j++)
        pthread_mutex_lock(&io_threads_mutex[j]);//
    server.io_threads_active = 0;
}
相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
8天前
|
NoSQL 安全 Unix
Redis源码、面试指南(4)单机数据库、持久化、通知与订阅(中)
Redis源码、面试指南(4)单机数据库、持久化、通知与订阅
18 0
|
2天前
|
缓存 NoSQL 中间件
【后端面经】【缓存】36|Redis 单线程:为什么 Redis 用单线程而 Memcached 用多线程?epoll、poll和select + Reactor模式
【5月更文挑战第19天】`epoll`、`poll`和`select`是Linux下多路复用IO的三种方式。`select`需要主动调用检查文件描述符,而`epoll`能实现回调,即使不调用`epoll_wait`也能处理就绪事件。`poll`与`select`类似,但支持更多文件描述符。面试时,重点讲解`epoll`的高效性和`Reactor`模式,该模式包括一个分发器和多个处理器,用于处理连接和读写事件。Redis采用单线程模型结合`epoll`的Reactor模式,确保高性能。在Redis 6.0后引入多线程,但基本原理保持不变。
21 2
|
3天前
|
缓存 NoSQL Redis
【后端面经】【缓存】36|Redis 单线程:为什么 Redis 用单线程而 Memcached 用多线程?--epoll调用和中断
【5月更文挑战第18天】`epoll`包含红黑树和就绪列表,用于高效管理文件描述符。关键系统调用有3个:`epoll_create()`创建epoll结构,`epoll_ctl()`添加/删除/修改文件描述符,`epoll_wait()`获取就绪文件描述符。`epoll_wait()`可设置超时时间(-1阻塞,0立即返回,正数等待指定时间)。当文件描述符满足条件(如数据到达)时,通过中断机制(如网卡或时钟中断)更新就绪列表,唤醒等待的进程。
32 6
|
4天前
|
NoSQL Redis 缓存
【后端面经】【缓存】36|Redis 单线程:为什么 Redis 用单线程而 Memcached 用多线程?
【5月更文挑战第17天】Redis常被称为单线程,但实际上其在处理命令时采用单线程,但在6.0后IO变为多线程。持久化和数据同步等任务由额外线程处理,因此严格来说Redis是多线程的。面试时需理解Redis的IO模型,如epoll和Reactor模式,以及其内存操作带来的高性能。Redis使用epoll进行高效文件描述符管理,实现高性能的网络IO。在讨论Redis与Memcached的线程模型差异时,应强调Redis的单线程模型如何通过内存操作和高效IO实现高性能。
33 7
【后端面经】【缓存】36|Redis 单线程:为什么 Redis 用单线程而 Memcached 用多线程?
|
6天前
|
NoSQL 算法 Java
【redis源码学习】持久化机制,java程序员面试算法宝典pdf
【redis源码学习】持久化机制,java程序员面试算法宝典pdf
|
6天前
|
Java
阅读《代码整洁之道》总结(1),java多线程面试
阅读《代码整洁之道》总结(1),java多线程面试
|
8天前
|
存储 NoSQL Redis
Redis源码、面试指南(5)多机数据库、复制、哨兵、集群(下)
Redis源码、面试指南(5)多机数据库、复制、哨兵、集群
224 1
|
8天前
|
监控 NoSQL Redis
Redis源码、面试指南(5)多机数据库、复制、哨兵、集群(上)
Redis源码、面试指南(5)多机数据库、复制、哨兵、集群
277 0
|
8天前
|
存储 NoSQL 调度
Redis源码、面试指南(4)单机数据库、持久化、通知与订阅(下)
Redis源码、面试指南(4)单机数据库、持久化、通知与订阅
12 0
|
8天前
|
存储 缓存 安全
Java 中 IO 流、File文件
Java 中 IO 流、File文件