Redis 网络库分析

文章目录

• • 1. 写在最前面
  • 2. I/O 多路复用
  • • 2.1 什么是 I/O 多路复用
    • 2.2 I/O 多路复用的实现
  • 3. 网络库源码结构
  • • 3.1 anet 文件
    • 3.2 ae 文件
    • • 3.2.1 aeEventLoop
      • 3.2.2 aeFileEvent
      • 3.2.3 aeTimeEvent
      • 3.2.4 aeFiredEvent
    • 3.3 ae_epoll、ae_select、kae_queue
  • 4. 思考
  • 5. 碎碎念
  • 6. 参考资料

1. 写在最前面

终于可以边调试边学习代码啦，其实调试的方法很简单，只需要三个步骤，值得拥有：

1. 下载源码 git clone https://github.com/redis/redis.git

2. checkout 到 1.3.6 的版本 git checkout -b 1.3.6 1.3.6

3. make，然后增加 debug 的 launch.json 配置文件如下：

{// Use IntelliSense to learn about possible attributes.// Hover to view descriptions of existing attributes.// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387"version": "0.2.0","configurations": [{"name": "clang++ - Build and debug active file","type": "cppdbg","request": "launch","program": "${workspaceFolder}/redis-server","args": [],"stopAtEntry": true,"cwd": "${workspaceFolder}","environment": [],"externalConsole": false,"MIMode": "lldb",// "preLaunchTask": "clang++ build active file"}]}

然后你就可以愉快的开始学习 redis 的源码啦^_

2. I/O 多路复用

在开始分析网络库之前，先来 redis 的网络库用到的背景知识—— I/O 多路复用。

2.1 什么是 I/O 多路复用

所谓「多路」指的是多个网络连接，「复用」指的是复用同一个线程。

注：采用多路 I/O 复用技术可以让单个线程高效的处理多个连接请求。

2.2 I/O 多路复用的实现

利用 select、epoll、kqueue 可以同时监听多个流的 I/O 事件的能力，在空闲的时候，会把当前线程阻塞掉，当有一个或多个流有 I/O 事件时，就从阻塞态中唤醒，于是程序就会轮询一遍所有的流，并且只依次顺序处理就绪的流。

3. 网络库源码结构

「不要依赖底层的顶层设计」这条原则，可以让你在重构代码的时候省去很多的麻烦，而 redis 网络库则是这条原则的「实践者」。整个 redis 的网络库部分大概由 10 个源码文件组成，且每类文件间功能划分的非常清晰。具体见下图：

3.1 anet 文件

提供的函数如下：

int anetTcpConnect(char *err, char *addr, int port);
int anetTcpNonBlockConnect(char *err, char *addr, int port);
int anetRead(int fd, char *buf, int count);
int anetResolve(char *err, char *host, char *ipbuf);
int anetTcpServer(char *err, int port, char *bindaddr);
int anetAccept(char *err, int serversock, char *ip, int *port);
int anetWrite(int fd, char *buf, int count);
int anetNonBlock(char *err, int fd);
int anetTcpNoDelay(char *err, int fd);
int anetTcpKeepAlive(char *err, int fd)

这层没有啥好说的，看函数名和入参能大概猜函数的用法，大致包含以下几类：

• 创建一个 tcp server —— anetTcpServer
• 接收 tcp 连接请求 —— anetAccept
• 连接 tcp server —— anetTcpConnect
• 读写网络 IO —— anetRead/anetWrite
• 设置网络 socket 的各种属性 —— anetTcpNoDelay/anetTcpKeepAlive/anetNonBlock

注：思考为什么需要 anetTcpNoDelay 的函数呢？

3.2 ae 文件

这两个文件是 redis 网络库抽象的核心，提供的函数和抽象的结构如下：

struct aeEventLoop;/* Types and data structures */
typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);
typedef void aeBeforeSleepProc(struct aeEventLoop *eventLoop);/* File event structure */
typedef struct aeFileEvent {int mask; /* one of AE_(READABLE|WRITABLE) */aeFileProc *rfileProc;aeFileProc *wfileProc;void *clientData;
} aeFileEvent;/* Time event structure */
typedef struct aeTimeEvent {long long id; /* time event identifier. */long when_sec; /* seconds */long when_ms; /* milliseconds */aeTimeProc *timeProc;aeEventFinalizerProc *finalizerProc;void *clientData;struct aeTimeEvent *next;
} aeTimeEvent;/* A fired event */
typedef struct aeFiredEvent {int fd;int mask;
} aeFiredEvent;/* State of an event based program */
typedef struct aeEventLoop {int maxfd;long long timeEventNextId;aeFileEvent events[AE_SETSIZE]; /* Registered events */aeFiredEvent fired[AE_SETSIZE]; /* Fired events */aeTimeEvent *timeEventHead;int stop;void *apidata; /* This is used for polling API specific data */aeBeforeSleepProc *beforesleep;
} aeEventLoop;/* Prototypes */
aeEventLoop *aeCreateEventLoop(void);
void aeDeleteEventLoop(aeEventLoop *eventLoop);
void aeStop(aeEventLoop *eventLoop);
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,aeFileProc *proc, void *clientData);
void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask);
long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,aeTimeProc *proc, void *clientData,aeEventFinalizerProc *finalizerProc);
int aeDeleteTimeEvent(aeEventLoop *eventLoop, long long id);
int aeProcessEvents(aeEventLoop *eventLoop, int flags);
int aeWait(int fd, int mask, long long milliseconds);
void aeMain(aeEventLoop *eventLoop);
char *aeGetApiName(void);
void aeSetBeforeSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *beforesleep);

3.2.1 aeEventLoop

管理文件事件（aeFileEvent）和时间事件（aeTimeEvent）的抽象结构。提供的函数具体包括

• aeCreateEventLoop —— 创建 aeEventLoop 结构

• aeCreateFileEvent/aeDeleteFileEvent —— 添加/删除 aeFileEvent 事件，并注册对应事件发生时的处理函数

• aeCreateTimeEvent/aeDeleteTimeEvent —— 添加/删除 aeTimeEvent 事件，并注册对应事件发生时的处理函数

• aeMain —— 在死循环中调用 aeProcessEvents（事件处理函数）检查，注册在 eventLoop 中的事件是否发生。

  void aeMain(aeEventLoop *eventLoop) {eventLoop->stop = 0;while (!eventLoop->stop) { // aeStop 退出循环检查if (eventLoop->beforesleep != NULL)eventLoop->beforesleep(eventLoop);aeProcessEvents(eventLoop, AE_ALL_EVENTS); // 调用 aeProcessEvent 检查注册的事件是否发生}
  }
  

• aeProcessEvents —— 处理 aeTimeEvent 和 aeFileEvent，先处理 aeFileEvent 后处理 aeTimeEvent

  int aeProcessEvents(aeEventLoop *eventLoop, int flags)
  {int processed = 0, numevents;if (eventLoop->maxfd != -1 ||((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {int j;aeTimeEvent *shortest = NULL;struct timeval tv, *tvp;// 计算离最近的 aeTimeEvent 到达还剩余多久shortest = aeSearchNearestTimer(eventLoop);// 如果 aeTimeEvent 还剩余指定时间，则可以阻塞指定时间// 如果 aeTimeEvent 已经到达，那就别等了，非阻塞查询numevents = aeApiPoll(eventLoop, tvp);for (j = 0; j < numevents; j++) {aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];int mask = eventLoop->fired[j].mask;int fd = eventLoop->fired[j].fd;int rfired = 0;/* note the fe->mask & mask & ... code: maybe an already processed* event removed an element that fired and we still didn't* processed, so we check if the event is still valid. */if (fe->mask & mask & AE_READABLE) {rfired = 1;fe->rfileProc(eventLoop,fd,fe->clientData,mask);}if (fe->mask & mask & AE_WRITABLE) {if (!rfired || fe->wfileProc != fe->rfileProc)fe->wfileProc(eventLoop,fd,fe->clientData,mask);}processed++;}}/* Check time events */if (flags & AE_TIME_EVENTS)processed += processTimeEvents(eventLoop);return processed; /* return the number of processed file/time events */
  }
  

3.2.2 aeFileEvent

抽象 server 与 client 建立的网络 socket 。对于 socket 上的读写事件，触发相应的处理操作。

3.2.3 aeTimeEvent

抽象各类基于时间的时间，包括周期性执行或者在给定时间点执行的事件。比如：

• 更新 server 的各类统计信息，比如时间、内存占用等
• 清理数据库中的过期键值对
• 关闭和清理连接失效的客户端
• 尝试进行 AOF 或者 RDB 持久化操作
• 等

3.2.4 aeFiredEvent

存储已经就绪的 fd 信息。

3.3 ae_epoll、ae_select、kae_queue

不管你底层的用的什么操作系统，底层用的 I/O 复用函数是什么，请务必封装出以下函数：

笔者所用的系统为 mac ，底层用的是 kqueue，所以总结下来其实就是封装 kqueque 函数的使用：

• aeApiCreate —— 封装 kqueue 函数，用于创建 fd

• aeApiPoll —— 封装 kevent 函数，用于检查就绪的 event

• aeApiAddEvent/aeApiDelEvent —— 封装 EV_SET 和 kevent 函数，用于添加和删除 event

kqueue 提供的函数介绍如下（ps 英文不好的小朋友，阔以看看下面的参考资料哦

SYNOPSIS: int kqueue(void);int kevent(int kq, const struct kevent *changelist, int nchanges,struct kevent *eventlist, int nevents,const struct timespec *timeout);EV_SET(&kev, ident, filter, flags, fflags, data, udata);
DESCRIPTION:The kqueue() system call creates a new kernel event queue and returns adescriptor. The kevent() system call is used to register events with the queue, andreturn any pending events to the user. The EV_SET() macro is provided for ease of initializing a kevent struc-ture. structure.ture.

4. 思考

1. redis 是基于顺序排队处理问题，那么对于大 key 和 value 这种情况，redis 的处理处理是否一样好呢？

   注： 大 key 会影响到计算 hash 的时间，大 value 会影响数据从磁盘拷贝到内存的时间等。

2. redis 的单线程模型和 nginx 的多进程模型真的只是因为作者任性嘛？

   redis 处理的数据是基于内存，内存特定是快

   nginx 在做反向代理的时候，处理的数据是需要跟后端进行大量的网络交换，时间 = 网络时间 + 业务逻辑处理时间 ，多数情况下会变慢

   产品的形态与所对应的业务场景有紧密的关联性，毕竟随着 redis 提供的能力越来越多的情况下，也开始支持多线程的版本

5. 碎碎念

嗯，写到这里突然就想起《天行九歌》的韩非子和《火影忍者》里的自来也了（ps 鬼知道，写个技术文章我怎么也会这么矫情。

• 最终，张良活成了韩非的模样 白凤活成了墨鸦的模样 只是，张良不喝酒 白凤不爱笑。 ——网易云《天行九歌》

• 自来也的人生总是差了一点，差一点挽回大蛇丸，差一点能触碰到纲手，差一点能把长门从黑暗里救出来，差一点就能看到鸣人的成长。 —— 网易云《自來也豪傑物語》

6. 参考资料

• 为什么说Redis是单线程的以及Redis为什么这么快！
• redis 网络通信模块设计与实现分析
• 什么是kqueue和IO复用
• Redis 源码分析

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！