redis 闲谈

深藏阁楼爱情的钟 2022-01-17 20:37 173阅读 0赞

### 看redis源码也有一阵子了 真要讲的话也 没什么大方向，就是简单的讲讲自己的理解 想一想就从大家经常用到的 Redis锁来作为起点来一步步讲解reids原理。 ###

## redis锁 ##

我们项目中经常用到分布式锁 最常用的就是 简单实用的redis锁  
下面是官方推荐写法: [redis.io/topics/dist…][redis.io_topics_dist]

上锁:  
SET key value NX PX 30000 (NX成功失败反馈,PX有效期)

原子脚本解锁 判断机器码 value 机器标识 防止误删:  
if redis.call("get",key) == val then  
  return redis.call("del",key)  
else  
  return 0  
end

官方把redis锁作为redlock的妥协性简单实现 那么它有什么不安全的？

单点问题--如果A获取到锁的并且还没有完成主从复制的瞬间 redis主挂了 主从备份是一般是异步操作 不保证完全不丢失数据 所以此时可能出现2个客户端同时获取到锁

## Redlock 官方推荐锁 ##

核心原理很简单 概括一下就是两点:  
1.N台redis服务器 客户端A请求对key加锁 当请求成功数n/2+1 超过半数 就认为成功 那么其他的客户端就无法在A释放前请求到n/2+1的reids 所以保证了只能有一个请求到锁  
2.如果每个客户端都没有请求到n/2+1 那么就设置随机时间错开去请求重试 直到请求成功为止  
3.一个非官方逻辑是:客户端要尽可能获取全部的锁，请求锁失败的客户端要释放所有的锁 然后再重新获取锁。

目前redlock 官方认可的java实现是redisson

#### 总结一下说说我个人的看法:    ####

Redlock 低概率异常

低概率一：超时,锁的有效期等计算 依赖多个服务器的时间一致

低概率二：如果有大量的客户端在请求锁 可能每个客户端都请求不到n/2+1 虽然redlock设置了随机延时 但是如果客户端量大 还是有几率一直获取不到锁

低概率三???：一共有3个Redis锁 A没有取到全部的锁取到了2个然后开始运行了 然后主挂了并且主从失败了 B这时候就能取到2个锁开始运行 如果Redis数量越多就概率越小

总体来说redlock因为有多个redis做保障，比单点出问题要小一些，虽然单点这种情况也是很极端  
个人认为RedLock 主要的弊端还是消耗资源比较大 为什么说消耗资源比较大 这要提到Reids高可用

目前高可用的方案主要是两种 单点+哨兵 和 集群

问题在于单点和集群都不能设置多个相同的KEY

所以实现RedLock的高可用需要多个单点+哨兵

## 单线程 ##

#### 因为Redis内存单线程设计保证了客户端操作有序 所以单单是它的get set就可以实现Redis锁 ####

### 单线程的Redis如何撑住大量访问 -- 从源码分析Redis事件模型 ###

在这之前我们先简单了解一下C语言的IO多路复用模型 我们以select模型为例

fd = socketconnect()  socket生成fd文件描述符
    FD_ZERO(fd_set*)      初始化fd_set集合
    FD_SET(fd,fd_set*)    将fd加入fd_set集合
    FD_CLR(fd,fd_set*)    将fd从fd_set集合删除
    FD_ISSET(fd,fd_set* ) 判断fd是否在fd_set集合
    
    int select(int maxfdp,fd_set *readfds,fd_set *writefds,fd_set *errorfds,struct timeval*timeout);
    
    maxfdp:fd_set中持有的最大的fd数量
    readfds:读fd集合 select执行后 会把触发的读事件留下其余的删除掉
    wreitefds:写fd集合 select执行后 会把触发的写事件留下其余的删除掉
    errorfds:错误fd集合
    timeval:超时 如果没有事件触发 那么等待timeval时间 然后超时结束 (这个超时对接下来的redis事件循环 起到了关键作用)
    复制代码

Redis的单线程运行原理就是围绕着上面的IO多路复用来运转的 接下来我们来用一张图来描述Redis的单线程运行方式

四个箭头内部是一个从左到右无限循环的结构，也是reids两大触发器: 文件触发器和时间触发器  
文件触发器就是IO多路复用触发器，时间触发器就是一个一段时间间隔运行一次的某个方法的触发器。  
从这可以看出redis有两种运行模式 一种是基于文件读写触发的运行，一种是按照时间节奏定时运行的,他们的主体都是基于一个无限循环的结构，接下来我们着重讲解文件事件。

### 带着问题来看文件事件与方法执行 ###

一、当IO模型选择出触发的fd以后那么他是如何执行方法的？

我们来看上图的注册器，注册器分为两部分 一部分是文件注册器，一部分是时间注册器，文件注册器将FD与读方法的函数指针和写方法的函数指针做了映射，当FD触发的时候 就可以根据读写标书去找到对应的方法来执行

二、fd对应一个socket连接 但是需要执行的方法有那么多 明显不对等 他们是怎么映射的？

这个问题我也纠结了很久，跟踪代码又很绕，很长一段时间都没有看出来，后来疯狂的编译源码打日志才看出来。  
可以看到我画的图里面一个FD分为读和写 也就是一个同时FD映射的方法有两个，读方法和写方法。  
**读方法**  
读方法对应了一个socket的数据解析 这个映射注册了以后一般是连接销毁后注销,他的触发就是对方发来数据后触发。（PS:我说解析数据的方法怎么那么长 那么多代码都堆在一起 原来是没有那么多FD可用??）  
**写方法**  
写方法虽然有很多，但是写方法是主动触发的，过程是 注册写方法->FD触发写方法->注销写方法 这种执行即销毁的方式让一个FD映射了多个方法

三、刚才提到主体是无限循环的结构 那么如何暂停？难道无限循环把cup跑满？

无限循环肯定需要sleep 要不然真把cup跑满了 这就要结合IO模型和时间事件说说了。  
在讲IO模型的时候我们提到了一个 timeval 超时阻塞了，就是这个超时阻塞控制了无限循环的节奏，当没有IO事件的时候最多阻塞timeval时间，这样保证了有IO事件立即执行，没有IO事件阻塞一段时间。

**那么阻塞多久那？**

先简单提一下时间触发器，举个例子 时间触发器里面有个触发方法 是100毫秒执行一次，那么如何保证在无限循环里面100毫秒执行一次？结合上面的timeval阻塞时间一下就应该明白了，在无限循环里，IO事件、时间事件是相互依赖的。 时间事件即保证了自己的执行又控制了无限循环的节奏和IO事件的超时。

#### 口说无凭 我们来看看代码 ####

先看一下关键的数据结构
    
    1.这个就是上面讲到的注册器结构体
    typedef struct aeEventLoop {
        aeFileEvent *events; 注册文件事件集合     
        aeFiredEvent *fired; IO触发后命中的FD集合 
        aeTimeEvent *timeEventHead;  时间事件集合
        int stop;   用来退出无限循环
        aeBeforeSleepProc *beforesleep; 循环前执行的方法
        aeBeforeSleepProc *aftersleep;  循环后执行的方法
        void *apidata; 这个是所有FD的总载体
     } aeEventLoop;
     //fired 和 events 都是集合  fired里面有fd events里面有函数指针和参数  他们两个是靠数组下表映射关系的
     
     
    2.注册器中的 FD的映射方法函数包装结构体 
    typedef struct aeFileEvent {
      int mask; 主要是READABLE|WRITABLE   读写标识
      aeFileProc *rfileProc;    读方法函数指针
      aeFileProc *wfileProc;   写方法函数指针
      void *clientData;   调用读写函数所传入的参数 读也传参数？读不是对方触发吗 传参数有什么意义？ 实际上读传的参数一般是对方的client 然后去取io buffer
    } aeFileEvent;
    
    3 命中后的Fd
    typedef struct aeFiredEvent {
        int fd;   //fd
        int mask; //读取标识
    } aeFiredEvent;
    
    4 存储FD用
    typedef struct aeApiState {
       fd_set rfds, wfds;  读FD集合 写FD集合
       fd_set _rfds, _wfds; 读写FD拷贝集合，因为SELECT会清空FD所以要传入拷贝集合
    }aeApiState;
    
    5 时间事件 
    typedef struct aeTimeEvent {
       long long id; 唯一标识id
       long when_sec; 秒
       long when_ms; 毫秒
       aeTimeProc *timeProc; 函数指针
       struct aeTimeEvent *prev;  链表结构
       struct aeTimeEvent *next;  链表结构
    } aeTimeEvent;
    
    
    接下来看看如何注册的
    
    
    //server.el 就是 aeEventLoop
    //link->fd 是对应的fd
    //clusterWriteHandler 是函数指针
    //link就是 aeEventLoop中的clientData 参数数据
    aeCreateFileEvent(server.el,link->fd,AE_WRITABLE,clusterWriteHandler,link){
        //把fd当做下标取出aeFileEvent
        aeFileEvent *fe = &eventLoop->events[fd];
        //设置fd
        aeApiAddEvent(eventLoop, fd, mask)
        //设置参数link
        fe->clientData = link;
        //设置读写函数指针 这个proc就是对应的clusterWriteHandler 方法
        if (mask & AE_READABLE) fe->rfileProc = proc;
        if (mask & AE_WRITABLE) fe->wfileProc = proc;
    }
    
    static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
         //把读写fd设置到一个总集合里面，因为IO多路复用每次调用都要重新设置fd 这个里面的fd就是用来每次取出来给IO复用的
         aeApiState *state = eventLoop->apidata;
         if (mask & AE_READABLE) FD_SET(fd,&state->rfds);
         if (mask & AE_WRITABLE) FD_SET(fd,&state->wfds);
         return 0;
    }
    
    
    接下来看看执行方法调用
    
    //server.c   redis入口main函数
    int main(int argc, char **argv){
        aeMain(server.el);//事件驱动入口
    }
    
    //ae.c 事件驱动入口
    void aeMain(aeEventLoop *eventLoop) {
       //无限循环
       while (!eventLoop->stop) {
         aeProcessEvents(eventLoop,AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
       }
    }
    
    //ae.c 事件处理器
    int aeProcessEvents(aeEventLoop *eventLoop, int flags){
            //这个算when_sec和when_ms 就是时间事件里面的分钟和秒 这一大堆就是算出tvp 给IO模型用
            long long ms =(shortest->when_sec - now_sec)*1000 +shortest->when_ms - now_ms;
            if (ms > 0) {
                tvp->tv_sec = ms/1000;
                tvp->tv_usec = (ms % 1000)*1000;
            } else {
                tvp->tv_sec = 0;
                tvp->tv_usec = 0;
            }
            
            //取出触发的IO事件,tvp就是上面根据时间事件算的阻塞时间
            numevents = aeApiPoll(eventLoop, tvp);
            
            //遍历取出来的IO触发事件根据数组下标找到映射
            for (j = 0; j < numevents; j++) {
                aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
                
                //如果是读事件调用rfileProc 函数指针 传递clientData 参数
                if (fe->mask & mask & AE_READABLE) {
                    fe->rfileProc(eventLoop,fd,fe->clientData,mask);
                    fired++;
                }
                //如果是写事件调用wfileProc 函数指针 传递clientData 参数
                if (fe->mask & mask & AE_WRITABLE) {
                  if (!fired || fe->wfileProc != fe->rfileProc) {
                    fe->wfileProc(eventLoop,fd,fe->clientData,mask);
                    fired++;
                  }
                }
            }
            
            //处理时间事件 在这不就具体讲时间事件了 
            if (flags & AE_TIME_EVENTS){
                processed += processTimeEvents(eventLoop);
            }
             
    
    }
    
    //ae_select.c 获取触发的IO事件
    int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
        //FD集合
        aeApiState *state = eventLoop->apidata;
        memcpy(&state->_rfds,&state->rfds,sizeof(fd_set));
        memcpy(&state->_wfds,&state->wfds,sizeof(fd_set));
        
        //SELECT IO 模型
        retval = select(eventLoop->maxfd+1,&state->_rfds,&state->_wfds,NULL,tvp);
        
        //读写命中取出FD
        if (fe->mask & AE_READABLE && FD_ISSET(j,&state->_rfds))
                mask |= AE_READABLE;
        if (fe->mask & AE_WRITABLE && FD_ISSET(j,&state->_wfds))
                mask |= AE_WRITABLE;
        eventLoop->fired[numevents].fd = j;
        eventLoop->fired[numevents].mask = mask;
        numevents++;
    }
    复制代码

### 小结 ###

通过这种方式，当有大量的客户端请求到来时 他们异步的进入IO模型，然后在通过IO选择器(select,poll...)单线程的批量执行，既保证了客户端的并发，有保证了内部简单有序，所以这种模式即便是简单的get set 方法 也可以用来当做redis锁。

# 待续...... #

## Redis主从复制 ##

### 上面提到Redis单点锁会在主从复制这个时间点有一定的危险性 那么接下来就简单介绍Redis的主从复制 因为网上主从复制的帖子有很多这里就不讲太细了 就挑几个点去讲 ###

//replication.c 这个文件承载了所有的主从复制
     //replicationCron 就是一个时间触发器来触发的方法
     replicationCron(){
        
        syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask);
        sendSynchronousCommand(SYNC_CMD_WRITE,fd,"PING",NULL);
        sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
        
     }
    复制代码

转载于:https://juejin.im/post/5cbebc455188250a8708d47a

[redis.io_topics_dist]: https://link.juejin.im?target=https%3A%2F%2Fredis.io%2Ftopics%2Fdistlock