redis线程池作用（redis集群三种方式）

redis 6引入多线程io，下面我们来和 netty 的多线程模型进行对比

分析思路：

• 初始化线程？
• 如何分配client给thread？
• 如何处理读写事件，在什么线程处理？
• 如何处理命令的逻辑，在什么线程处理？

netty的多线程模型

用户代码

serverbootstrap b = new serverbootstrap();
            b.group(bossgroup, workergroup)
            .channel(nioserversocketchannel.class)
            .childoption(channeloption.tcp_nodelay, true)
            .childattr(attributekey.newinstance("childattr"), "childattrvalue")
            .handler(new serverhandler())
            .childhandler(new channelinitializer<socketchannel>() {
             @override
             public void initchannel(socketchannel ch) {
              ch.pipeline().addlast(new authhandler());
               //..}});

             channelfuture f = b.bind(8888).sync();
             f.channel().closefuture().sync();
复制代码

初始化线程( serverbootsrap.bind())

netty初始化线程，创建一个 boss线程池，一个work线程池，并且给new了一个channel处理注册线程的连接，并且为这个channel 添加了一个 serverbootstrapacceptor的channel。

• 操作线程： 主线程执行
• 执行时机： 初始化线程
• 执行代码： serverbootsrap.bind()

如何分配 client给thread？

• 操作线程 ： 主线程执行
• 执行时机 ： 新连接接入

新连接的建立可以分为三个步骤 1.检测到有新的连接 2.将新连接注册到work线程组 3.注册新连接的读事件

boss线程组的 nioeventloop.run() 不断检查所有的管道，当管道状态为可读或者连接的时候就会读取管道。

if ((readyops & (selectionkey.op_read | selectionkey.op_accept)) != 0 || readyops == 0) {
    unsafe.read();
}
复制代码

然后就按 channel 的责任链传递下去

• unsafe.read()
• —->
• pipeline.firechannelread(bytebuf);
• —->
• serverbootstrapacceptor.channelread()
• —->
• multithreadeventloopgroup.register(child) 分配一个线程给这个channel，一个线程可能拥有多个channel

如何匹配线程

defaulteventexecutorchooserfactory.java 
用线程个数取余来分配
@override
public eventexecutor next() {
    return executors[math.abs(idx.getandincrement() % executors.length)];
}



abstractchannel.java
@override
public final void register(eventloop eventloop, final channelpromise promise) {
    // 重点！！！ 这个线程就永远被挂靠在channel上面了
    abstractchannel.this.eventloop = eventloop;
    // 监听读事件 nio底层的注册
    register0(promise);
    }
}
复制代码

如何处理读写事件，在什么线程处理？

channelinboundhandler.channelread

如何处理命令的逻辑，在什么线程处理？

channelinboundhandler.channelread

总结：netty通过开始注册一个boss线程池（通常情况都是一个），来监听(nioeventloop,run)连接的channel，如果有channel来进行连接，就通过责任链找到
serverbootstrapacceptor.channelread() 分配给channel一个线程（nioeventloop），这个线程（nioeventloop）就会通过run()去不断的 去读channel里面的，处理命令。

redis的多线程模型

初始化线程( initthreadedio() 函数)

• 操作线程 ：主线程执行
• 执行时机 ：初始化线程

首先，如果用户没有开启多线程io，也就是 io_thread_num ==1，按照单线程处理； 如果超过线程数io_threads_max_num上限则异常退出。

创建io_threads_num个线程（listcreate），并且对除主线程（id==0）以外的线程进行处理： （listcreate 创建一个线程）

1. 初始化线程的等待任务为0
2. 获取锁，使得线程不能进行操作
3. 将线程tid与redis中的线程id进行映射

/* initialize the data structures needed for threaded i/o. */
void initthreadedio(void) {
    io_threads_active = 0; /* we start with threads not active. */
 
    /* don't spawn any thread if the user selected a single thread:
     * we'll handle i/o directly from the main thread. */
    // 如果用户没有开启多线程io直接返回 使用主线程处理
    if (server.io_threads_num == 1) return;
    // 线程数设置超过上限
    if (server.io_threads_num > io_threads_max_num) {
        serverlog(ll_warning,"fatal: too many i/o threads configured. "
                             "the maximum number is %d.", io_threads_max_num);
        exit(1);
    }
 
    /* spawn and initialize the i/o threads. */
    // 初始化io_threads_num个对应线程
    for (int i = 0; i < server.io_threads_num; i++) {
        /* things we do for all the threads including the main thread. */
        io_threads_list[i] = listcreate();
        if (i == 0) continue; // index 0为主线程，跳过
 
        /* things we do only for the additional threads. */
        // 非主线程则需要以下处理
        pthread_t tid;
        // 为线程初始化生成对应的锁
        pthread_mutex_init(&io_threads_mutex[i],null);
        // 线程等待状态初始化为0
        io_threads_pending[i] = 0;
         // 初始化后将线程锁住
        pthread_mutex_lock(&io_threads_mutex[i]);
        if (pthread_create(&tid,null,iothreadmain,(void*)(long)i) != 0) {
            serverlog(ll_warning,"fatal: can't initialize io thread.");
            exit(1);
        }
        // 将index和对应线程id加以映射
        io_threads[i] = tid;
    }
}

复制代码

读事件到来（readqueryfromclient）

• 操作线程 ：主线程执行
• 机制时机 ：读事件到来

redis需要判断是否满足 thread io 条件，执行 postponeclientread，执行后会将 client放到等待读取的队列，并将client设置为等待读取的状态。

// 读取到一个客户端的请求
int postponeclientread(client *c) {
    if (io_threads_active && // 线程是否在不断(spining)等待io
        server.io_threads_do_reads && // 是否多线程io读取
        !(c->flags & (client_master|client_slave|client_pending_read)))
    {//client不能是主从，且未处于等待读取的状态
    // 将client设置为等待读取的状态flag
        c->flags |= client_pending_read; 
        // 把client加入到等待读列表
        listaddnodehead(server.clients_pending_read,c); 
        return 1;
    } else {
        return 0;
    }
}
复制代码

这时server维护了一个 clients_pending_read，包含所有的读事件 pending的客户端列表。

如何分配client给thread(线程) （handleclientswithpendingreadsusingthreads）

• 操作线程 ：主线程执行
• 执行时机 ：执行处理事件之后

首先，redis检查有等待的读client listlength(
server.clients_pending_read)

如果是长度不为0，进行while循环，将每个等待的client分配给线程，当等待长度超过线程时， 每个线程分配给到的client可能超过1个；

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++;
}

并且修改每个线程需要完成的数量（初始化为0）：
// 所有线程
for (int j = 1; j < server.io_threads_num; j++) {
    // 拿出当前线程需要处理多少个客户端
    int count = listlength(io_threads_list[j]);
    // 设置当前线程需要多少客户端
    io_threads_pending[j] = count;
}


等待处理直到没有剩余任务：
while(1) {
    unsigned long pending = 0;
        // 拿出所有线程，查看线程是否还有需要的客户端
        // 这里主要是监听子线程是否完全处理好任务
        for (int j = 1; j < server.io_threads_num; j++)
        pending += io_threads_pending[j];
    if (pending == 0) break;
}

当本次io的所有(读/写)线程处理完毕之后，清空client_pending_read:
主线程会在这里处理命令
listrewind(server.clients_pending_read,&li);
while((ln = listnext(&li))) {
    client *c = listnodevalue(ln);
    c->flags &= ~client_pending_read;
    if (c->flags & client_pending_command) {
        c->flags &= ~ client_pending_command;
        processcommandandresetclient(c);
    }
    processinputbufferandreplicate(c);
}
listempty(server.clients_pending_read);
复制代码

如何处理读请求（iothreadmain）

• 操作线程 ：子线程
• 执行时机 ： 子线程启动时 while执行

在上面过程中，当任务分发完毕后，每个线程按照正常流程将自己负责的client的读取缓冲区的内容进行处理，和原来的单线程没有太大差异。

redis为每个客户端分配了输入缓冲区，它的作用是将客户端发送的命令临时保存，同时redis从会输入缓冲区拉取命令并执行，输入缓冲区为客户端发送命令到redis执行命令提供了缓冲功能。

redis的 thread io 模型中，每次所有的线程都只能进行或者 写/读 操作，通过 io_threads_op控制。 同时每个线程负责的client一次执行：

// io thread主函数，在各个子线程执行
void *iothreadmain(void *myid) {
    // 线程 id，跟普通线程池的操作方式一样，都是通过 线程id 进行操作
    long id = (unsigned long)myid;
    while(1) {
        /* 
        *这里的等待操作比较特殊，没有使用简单的 sleep，
        *避免了 sleep 时间设置不当可能导致糟糕的性能，
        *但是也有个问题就是频繁 loop 可能一定程度上造成 cpu 占用较长
         */
        for (int j = 0; j < 1000000; j++) {
            if (io_threads_pending[id] != 0) break;
        }
        // 根据线程 id 以及待分配列表进行 任务分配
        listiter li;
        listnode *ln;
        listrewind(io_threads_list[id],&li);
        // 有可能分配了两个客户端连接
        while((ln = listnext(&li))) {
            client *c = listnodevalue(ln);
            if (io_threads_op == io_threads_op_write) {
                 // 当前全局处于写事件时，向输出缓冲区写入响应内容
                writetoclient(c,0);
            } else if (io_threads_op == io_threads_op_read) {
                // 当前全局处于读事件时，从输入缓冲区读取请求内容
                readqueryfromclient(c->conn);
            } else {
                serverpanic("io_threads_op value is unknown");
            }
        }
        listempty(io_threads_list[id]);
        io_threads_pending[id] = 0;
        if (tio_debug) printf("[%ld] donen", id);
    }

复制代码

readqeuryfromclient()->processinputbuffer(c)->processcommand() 进行command的分发和处理。

这里的readqueuefromclient只做写入客户端的输入缓存区：

// 复制到 client 缓存区
else if (c->flags & client_master) {
        c->pending_querybuf = sdscatlen(c->pending_querybuf,
                                        c->querybuf+qblen,nread);
    }
void processinputbuffer(client *c) {
     while(c->qb_pos < sdslen(c->querybuf)) {
         
         
        // 如果我们在 io线程（子线程）的时候
        // 我们不能直接执行命令，flags设置为client_pending_command
        // 然后让主线程执行
        if (c->flags & client_pending_read) {
            c->flags |= client_pending_command;
            break;
            }
            }
}
复制代码

每个线程执行readqueryfromclient，将对应的请求放入一个队列，单线程执行（从输入缓存区读取内容），线程将结果写入客户端的buff中。

每轮处理中，需要将各个线程的锁开启，并且将相关标志位：

void startthreadedio(void) {
    if (tio_debug) { printf("s"); fflush(stdout); }
    if (tio_debug) printf("--- starting threaded io ---n");
    serverassert(io_threads_active == 0);
    for (int j = 1; j < server.io_threads_num; j++)
        // 解开线程的锁定状态
        pthread_mutex_unlock(&io_threads_mutex[j]);
    // 现在可以开始多线程io执行对应读/写任务
    io_threads_active = 1;
}

复制代码

结束时，首先检查是否有是否有待读的io，如果没有，将线程说的，标志关闭：

void stopthreadedio(void) {
    // 需要停止的时候可能还有等待读的client 在停止前进行处理
    handleclientswithpendingreadsusingthreads();
    if (tio_debug) { printf("e"); fflush(stdout); }
    if (tio_debug) printf("--- stopping threaded io [r%d] [w%d] ---n",
        (int) listlength(server.clients_pending_read),
        (int) listlength(server.clients_pending_write));
    serverassert(io_threads_active == 1);
    for (int j = 1; j < server.io_threads_num; j++)
        // 本轮io结束 将所有线程上锁
        pthread_mutex_lock(&io_threads_mutex[j]);
    // io状态设置为关闭
    io_threads_active = 0;
}
复制代码

总结：threaded io将服务读client的输入缓存区和将执行结果写入输出缓冲区的过程改为了多线程模型， 同时保持同一时间全部线程均处于读或写的状态。但是命令的具体执行以单线程（队列）的形式。 因为redis希望保持坚定结果避免锁和竞争问题，并且读写缓存占用命令执行声明周期的比重比大 ，处理这部分io模型给性能带来来显著的提升。

netty 和 redis6 区别：

如何分配client给thread？

netty：当boss监听到连接事件，netty会给一个channel分配一个线程。这个线程专门负责这条channel的读写事件，可以是解析也可以是执行命令

redis6：每当接收到一个读事件，client放到等待读取的队列。在执行处理事件之后，主线程会 统一给线程池的线程分配client。线程把client要读buffer都放到client的缓存。主线程等待所有 io线程执行完毕，主线程再执行client的缓存变成命令

如何处理读写事件，在什么线程处理？

netty：在子线程执行，读写完直接执行逻辑 redis6：在子线程执行，读写完放在client的缓冲区

如何处理命令的逻辑，在什么线程处理？

netty：在子线程执行，直接执行逻辑 redis6：在主线程执行，主线程遍历等待队列读取缓冲区编译成命令再执行

为什么redis选择使用这个模式？