从源码出发搞懂redis网络,你必须知道的一些事
liebian365 2024-10-27 13:14 7 浏览 0 评论
本篇文章将讲一下redis的网络架构,来揭秘redis接收数据的全过程
基础网络知识
再具体研究redis的代码之前,我们得首先得有一个大体的服务器如何接收到客户端数据的全过程。
因为redis 主要是用tcp在传输数据,所以是四层网络。
那么具体在应用层,我们又如何通过套接字接收一个客户端的数据了。
请看以下代码分析
redis的监听于地址绑定
server.c
/* Open the TCP listening socket for the user commands. */ //普通tcp的监听启动 if (server.port != 0 && //server.ipfd ip绑定地址 ipfd 数组长度为16,每一项表示一个绑定地址对应的句柄。 //ipfd_count 这里指的是监听地址的个数,具体跟你配置的地址有关系,一般我们配置127.0.0.1 那么他就是只会去和本机的服务进行通信 //ipfd 是一个整型数组,在这个方法里面会被赋值,ipfd[0] 等于文件描述符号,也可以理解为socket的id号 listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR) exit(1); //tls的监听启动 if (server.tls_port != 0 && listenToPort(server.tls_port,server.tlsfd,&server.tlsfd_count) == C_ERR) exit(1);
/* Open the listening Unix domain socket. */ //适配unix if (server.unixsocket != NULL) { unlink(server.unixsocket); /* don't care if this fails */ server.sofd = anetUnixServer(server.neterr,server.unixsocket, server.unixsocketperm, server.tcp_backlog); if (server.sofd == ANET_ERR) { serverLog(LL_WARNING, "Opening Unix socket: %s", server.neterr); exit(1); } anetNonBlock(NULL,server.sofd); }
我们拿linux系统做为例子
- 要想一个客户端访问到服务器首先我们得使用socket 开放一个端口,告诉服务器现在这个端口再监听,当客户端访问该端口的时候,socket对应的文件句柄状态会发生改变。
- 而每个端口又可以绑定多个地址,比如 127.0.0.1 我们只允许本地访问,比如0.0.0.0 表示所有ip都可以访问,等等。默认情况下redis 是用的0.0.0.0.
从第一个listenToPort进入后的代码如下:
server.c
int listenToPort(int port, int *fds, int *count) {
int j;
/* Force binding of 0.0.0.0 if no bind address is specified, always
* entering the loop if j == 0. */
//强制绑定 0.0.0.0 表示任何ip 可否问
if (server.bindaddr_count == 0) server.bindaddr[0] = NULL;
//可以看到绑定的地址是一个集合
for (j = 0; j < server.bindaddr_count || j == 0; j++) {
if (server.bindaddr[j] == NULL) {
int unsupported = 0;
/* Bind * for both IPv6 and IPv4, we enter here only if
* server.bindaddr_count == 0. */
//如果绑定的地址为空,则初始化 ipv6 和 ipv4
//第一个是错误网络信息,提供的一个buffer
//现实启动绑定ipv6的socket
//fds 是一个整型数组, fds[*count] 表示这个地址下的,文件描述符
//server.tcp_backlog 指的是tcp协议里面,三次握手里面,服务端在tcp
//握手的第三个阶段完成后,会将这些standby的
//连接放入这个accept queue 里面
//anetTcp6Server和anetTcpServer 会给这个监听的地址和端口分配一个fd
fds[*count] = anetTcp6Server(server.neterr,port,NULL,
server.tcp_backlog);
if (fds[*count] != ANET_ERR) {
//设置为非阻塞
//fds[*count] 应该等于设置的 server.tcp_backlog
anetNonBlock(NULL,fds[*count]);
(*count)++;
} else if (errno == EAFNOSUPPORT) {
unsupported++;
serverLog(LL_WARNING,"Not listening to IPv6: unsupported");
}
if (*count == 1 || unsupported) {
/* Bind the IPv4 address as well. */
//下面的逻辑相同
fds[*count] = anetTcpServer(server.neterr,port,NULL,
server.tcp_backlog);
if (fds[*count] != ANET_ERR) {
anetNonBlock(NULL,fds[*count]);
(*count)++;
} else if (errno == EAFNOSUPPORT) {
unsupported++;
serverLog(LL_WARNING,"Not listening to IPv4: unsupported");
}
}
/* Exit the loop if we were able to bind * on IPv4 and IPv6,
* otherwise fds[*count] will be ANET_ERR and we'll print an
* error and return to the caller with an error. */
if (*count + unsupported == 2) break;
}
//如果是ipv6的地址类型,单独做ipv6地址的监听
else if (strchr(server.bindaddr[j],':')) {
/* Bind IPv6 address. */
fds[*count] = anetTcp6Server(server.neterr,port,server.bindaddr[j],
server.tcp_backlog);
} else {
/* Bind IPv4 address. */
// ipv4的地址的绑定
fds[*count] = anetTcpServer(server.neterr,port,server.bindaddr[j],
server.tcp_backlog);
}
if (fds[*count] == ANET_ERR) {
//这里是打印日志的地方
serverLog(LL_WARNING,
"Could not create server TCP listening socket %s:%d: %s",
server.bindaddr[j] ? server.bindaddr[j] : "*",
port, server.neterr);
if (errno == ENOPROTOOPT || errno == EPROTONOSUPPORT ||
errno == ESOCKTNOSUPPORT || errno == EPFNOSUPPORT ||
errno == EAFNOSUPPORT || errno == EADDRNOTAVAIL)
continue;
return C_ERR;
}
anetNonBlock(NULL,fds[*count]);
(*count)++;
}
return C_OK;
}
anet.c
static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backlog)
{
int s = -1, rv;
//最高的端口不超过65535
char _port[6]; /* strlen("65535") */
struct addrinfo hints, *servinfo, *p;
snprintf(_port,6,"%d",port);
memset(&hints,0,sizeof(hints));
//ipv6或者ipv4 两者主要区别在于,地址的位数不同,且格式也不同,
//本质的用处都一样
hints.ai_family = af;
//流式socket 即tcp 传输协议
hints.ai_socktype = SOCK_STREAM;
//获取地址,即当bindaddr为null的时候,返回0.0.0.0
hints.ai_flags = AI_PASSIVE; /* No effect if bindaddr != NULL */
//IPv4中使用gethostbyname()函数完成主机名到地址解析,这个函数仅仅支持IPv4,且不允许调用者指定所需地址类型的任何信息,返回的结构只包含了用于存储IPv4地址的空间。
// IPv6中引入了getaddrinfo()的新API,它是协议无关的,既可用于IPv4也可用于IPv6。
// getaddrinfo函数能够处理名字到地址以及服务到端口这两种转换,返回的是一个addrinfo的结构(列表)指针而不是一个地址清单。
// 这些addrinfo结构随后可由套接口函数直接使用。如此以来,getaddrinfo函数把协议相关性安全隐藏在这个库函数内部。
// 应用程序只要处理由getaddrinfo函数填写的套接口地址结构。该函数在 POSIX规范中定义了。
if ((rv = getaddrinfo(bindaddr,_port,&hints,&servinfo)) != 0) {
anetSetError(err, "%s", gai_strerror(rv));
return ANET_ERR;
}
for (p = servinfo; p != NULL; p = p->ai_next) {
//如果socket 初始化不成功则跳过,比如端口被占用
if ((s = socket(p->ai_family,p->ai_socktype,p->ai_protocol)) == -1)
continue;
//如果ipv6初始化失败 而且仅仅只初始化ipv6 则进入error
if (af == AF_INET6 && anetV6Only(err,s) == ANET_ERR) goto error;
//设置地址复用,地址复用的意思就是可以允许进程复用端口
if (anetSetReuseAddr(err,s) == ANET_ERR) goto error;
//绑定地址,初始化accept queue的长度
if (anetListen(err,s,p->ai_addr,p->ai_addrlen,backlog) == ANET_ERR) s = ANET_ERR;
goto end;
}
if (p == NULL) {
anetSetError(err, "unable to bind socket, errno: %d", errno);
goto error;
}
error:
if (s != -1) close(s);
s = ANET_ERR;
end:
freeaddrinfo(servinfo);
return s;
}
上面的代码就是一个socket 初始化的过程,
- 获取一个socket 可认识的地址结构。
- 然后调用socket方法,声明为tcp的网络模式还是udp的网络模式,返回得到一个文件描述符。
- 设置绑定的地址,每个绑定地址都会对应一个fd(文件描述符),fd的表现形式就是一个int型的非负整数。这边要注意的是backlog这个参数,在上面代码给了比较详细的注释,backlog在并发比较高的状态下建议可以设置大一点。它一般和系统变量/proc/sys/net/core/somaxconn一起使用,两者取其小。
上面的整个过程就是,redis socket初始化的过程,下面要开始说redis socket 的监听环节。
推荐视频:
为什么C/C++程序员一定要阅读redis源码?腾讯面试教你做人
学习地址:C/C++Linux服务器开发/后台架构师【零声教育】-学习视频教程-腾讯课堂
需要更多C/C++ Linux服务器架构师学习资料加群812855908(资料包括C/C++,Linux,golang技术,内核,Nginx,ZeroMQ,MySQL,Redis,fastdfs,MongoDB,ZK,流媒体,CDN,P2P,K8S,Docker,TCP/IP,协程,DPDK,ffmpeg,大厂面试题 等)
redis 的网络监听事件配置
上面的步骤可以让客户端发动信息到,redis绑定的端口上来了,但是连接千千万万,我们又如何采用合适的策略知道那些现在需要被处理的连接了,那就需要配置socket的相关监听了。
server.c
//这里开始对我们监听的地址分配handler 事件
//一般情况下ipfd_count=1
for (j = 0; j < server.ipfd_count; j++) {
//为这个socket创建一个只读事件
//并且为事件处理准备了一个handler,就是acceptTcpHandler
if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
acceptTcpHandler,NULL) == AE_ERR)
{
serverPanic(
"Unrecoverable error creating server.ipfd file event.");
}
}
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
aeFileProc *proc, void *clientData)
{
//fd 可以看成一个自增数,
if (fd >= eventLoop->setsize) {
errno = ERANGE;
return AE_ERR;
}
//创建一个file event的引用
aeFileEvent *fe = &eventLoop->events[fd];
//将fd 放入到io监听事件里面去
if (aeApiAddEvent(eventLoop, fd, mask) == -1)
return AE_ERR;
fe->mask |= mask;
//指定读handler和写handler
if (mask & AE_READABLE) fe->rfileProc = proc;
if (mask & AE_WRITABLE) fe->wfileProc = proc;
//给予一个客户端data 的引用
fe->clientData = clientData;
if (fd > eventLoop->maxfd)
//更新当前最大文件描述符
eventLoop->maxfd = fd;
return AE_OK;
}
/* Include the best multiplexing layer supported by this system.
* The following should be ordered by performances, descending. */
// 根据不同的系统 选用不同的io实现, 以下实现的性能按照上下排序, 1, ae_evport.c ,2 ae_epoll.c ,3 ae_kqueue.c ,4,ae_select.c
// 在linux 系统会优先用到epoll
#ifdef HAVE_EVPORT
#include "ae_evport.c"
#else
#ifdef HAVE_EPOLL
#include "ae_epoll.c"
#else
#ifdef HAVE_KQUEUE
#include "ae_kqueue.c"
#else
#include "ae_select.c"
#endif
#endif
#endif
这里面的知识点会比较多,比如epoll 的工作机制是如何,select 工作机制又是如何,但是这里重点讲的使用整个过程
可以看到这里我们会把刚刚的socket 相关的文件描述符放入到整个io事件监听体系里面去,在Linux 系统里面会默认用到epoll 这个模型,这里稍微再介绍下ae_epoll.c里面的几个方法
typedef struct aeApiState {
int epfd;
struct epoll_event *events;
} aeApiState;
// redis epoll的初始化
static int aeApiCreate(aeEventLoop *eventLoop) {
// epoll 模型结构分配空间
aeApiState *state = zmalloc(sizeof(aeApiState));
if (!state) return -1;
//根据setSize 分配事件空间,一个事件对应一个fd
state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);
if (!state->events) {
zfree(state);
return -1;
}
//创建一个epoll的模型, 1024 这个?代表的epoll 可能会处理的fd 数目,高核版本这个参数没有什么意义,epoll创建也会得到一个文件描述符,表示打开了一个文件,所以当不用的时候也是需要对其进行关闭。
state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */
if (state->epfd == -1) {
zfree(state->events);
zfree(state);
return -1;
}
eventLoop->apidata = state;
return 0;
}
//将目标fd的读写事件注册到epoll里面,让epoll的线程帮助那些可读可写的fd准备好,而不需要主线程阻塞式的去寻找。
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
aeApiState *state = eventLoop->apidata;
struct epoll_event ee = {0}; /* avoid valgrind warning */
/* If the fd was already monitored for some event, we need a MOD
* operation. Otherwise we need an ADD operation. */
//判断当前文件描述符 是否已经被创建
int op = eventLoop->events[fd].mask == AE_NONE ?
EPOLL_CTL_ADD : EPOLL_CTL_MOD;
ee.events = 0;
mask |= eventLoop->events[fd].mask; /* Merge old events */
if (mask & AE_READABLE) ee.events |= EPOLLIN;
if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
ee.data.fd = fd;
//为这个fd 注册 epollin 或者 epollout 事件,如果之前已经创建过,则增加事件类型,正因为有修改和删除的操作,epoll 内部维护了一棵红黑树使得增加,修改,删除的事件复杂度为logn
if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
return 0;
}
//删除的逻辑,当可读,可写事件都存在的时候,删除操作也可以看作是一种修改。
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) {
aeApiState *state = eventLoop->apidata;
struct epoll_event ee = {0}; /* avoid valgrind warning */
int mask = eventLoop->events[fd].mask & (~delmask);
ee.events = 0;
if (mask & AE_READABLE) ee.events |= EPOLLIN;
if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
ee.data.fd = fd;
if (mask != AE_NONE) {
//如果只是消除可读可写的事件某一件的时候那就是修改
epoll_ctl(state->epfd,EPOLL_CTL_MOD,fd,&ee);
} else {
/* Note, Kernel < 2.6.9 requires a non null event pointer even for
* EPOLL_CTL_DEL. */
epoll_ctl(state->epfd,EPOLL_CTL_DEL,fd,&ee);
}
}
//通过epoll wait 拉到对应的可被读写的fd
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
aeApiState *state = eventLoop->apidata;
int retval, numevents = 0;
//通过这个方法拉取到可读可写的fd,
//第二个参数是拉取个中状态的事件,记住可以拉多种状态一次性
//第三个参数拉取事件最大数,一个fd对应一个事件,一个事件可以为即可读又可写的状态
//第四个参数为超时时间。-1 为一直等待。
retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
if (retval > 0) {
int j;
numevents = retval;
for (j = 0; j < numevents; j++) {
int mask = 0;
struct epoll_event *e = state->events+j;
//可以看到c语言常用位数来表示状态,这样的话多状态就可以表现出来了。
if (e->events & EPOLLIN) mask |= AE_READABLE;
if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
if (e->events & EPOLLERR) mask |= AE_WRITABLE|AE_READABLE;
if (e->events & EPOLLHUP) mask |= AE_WRITABLE|AE_READABLE;
eventLoop->fired[j].fd = e->data.fd;
eventLoop->fired[j].mask = mask;
}
}
return numevents;
}
epoll的使用总结
首先要明白epoll不是具体去与客户端的沟通的媒介, 而是一种对fd的监听架构,它主要运作在系统内核里面,当其监听下的fd的状态发生改变时,根据fd的注册事件类型,返回对应的fd,告诉主程序该fd需要处理。所以总结起来使用epoll 也非常简单,
1.初始化的时候,使用epoll_create
2.注册事件的时候,使用epoll_ctl
3. 拉取有状态改变的fd ,使用epoll_wait
events可以是以下几个宏的集合:
EPOLLIN :表示对应的文件描述符可以读(包括对端SOCKET正常关闭);
EPOLLOUT:表示对应的文件描述符可以写;
EPOLLPRI:表示对应的文件描述符有紧急的数据可读(这里应该表示有带外数据到来);
EPOLLERR:表示对应的文件描述符发生错误;
EPOLLHUP:表示对应的文件描述符被挂断;
EPOLLET: 将EPOLL设为边缘触发(Edge Triggered)模式,这是相对于水平触发(Level Triggered)来说的。
EPOLLONESHOT:只监听一次事件,当监听完这次事件之后,如果还需要继续监听这个socket的话,需要再次把这个socket加入到EPOLL队列里
为了避免对redis代码理解引入新的问题,我们需要着重关注两个状态,即在redis里用到的EPOLLIN 和 EPOLLOUT。
其实看完上面的代码之后我们一定会有一个疑问,我们只注册了一个(默认情况下只有一个)fd到epoll 里面去,那我们为什么还要使用epoll了,下面继续分析
redis 的网络监听事件
首先我们回到server.c的代码
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
//这里进入死循环
while (!eventLoop->stop) {
//启动事件
aeProcessEvents(eventLoop, AE_ALL_EVENTS|
AE_CALL_BEFORE_SLEEP|
AE_CALL_AFTER_SLEEP);
}
}
可以看到上面进入aeMain 这个方法之后,主线程会循环调用处理事件方法。
/* Process every pending time event, then every pending file event
* (that may be registered by time event callbacks just processed).
* Without special flags the function sleeps until some file event
* fires, or when the next time event occurs (if any).
*
* If flags is 0, the function does nothing and returns.
* if flags has AE_ALL_EVENTS set, all the kind of events are processed.
* if flags has AE_FILE_EVENTS set, file events are processed.
* if flags has AE_TIME_EVENTS set, time events are processed.
* if flags has AE_DONT_WAIT set the function returns ASAP until all
* the events that's possible to process without to wait are processed.
* if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called.
* if flags has AE_CALL_BEFORE_SLEEP set, the beforesleep callback is called.
*
* The function returns the number of events processed. */
// 可以看到redis 是一个事件驱动型的设计模式,会循环调用时间事件
//(如cycle 之前提到的过期键的处理) 然后还有文件事件(其实就是处理客户端的请求),
// 每次循环完还有一些回调的策略方法放在我们的beforesleep,aftersleep 里面。
// 下面这段代码,我们首先关注文件型事件。
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
int processed = 0, numevents;
/* Nothing to do? return ASAP */
if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
/* Note that we want call select() even if there are no
* file events to process as long as we want to process time
* events, in order to sleep until the next time event is ready
* to fire. */
if (eventLoop->maxfd != -1 ||
((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
int j;
aeTimeEvent *shortest = NULL;
struct timeval tv, *tvp;
if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
shortest = aeSearchNearestTimer(eventLoop);
if (shortest) {
long now_sec, now_ms;
aeGetTime(&now_sec, &now_ms);
tvp = &tv;
/* How many milliseconds we need to wait for the next
* time event to fire? */
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;
}
} else {
/* If we have to check for events but need to return
* ASAP because of AE_DONT_WAIT we need to set the timeout
* to zero */
if (flags & AE_DONT_WAIT) {
tv.tv_sec = tv.tv_usec = 0;
tvp = &tv;
} else {
/* Otherwise we can block */
tvp = NULL; /* wait forever */
}
}
if (eventLoop->flags & AE_DONT_WAIT) {
tv.tv_sec = tv.tv_usec = 0;
tvp = &tv;
}
if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)
eventLoop->beforesleep(eventLoop);
/* Call the multiplexing API, will return only on timeout or when
* some event fires. */
// 获取事件个数
numevents = aeApiPoll(eventLoop, tvp);
/* After sleep callback. */
if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
eventLoop->aftersleep(eventLoop);
//这里是开始处理网络请求的地方。
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 fired = 0; /* Number of events fired for current fd. */
/* Normally we execute the readable event first, and the writable
* event laster. This is useful as sometimes we may be able
* to serve the reply of a query immediately after processing the
* query.
*
* However if AE_BARRIER is set in the mask, our application is
* asking us to do the reverse: never fire the writable event
* after the readable. In such a case, we invert the calls.
* This is useful when, for instance, we want to do things
* in the beforeSleep() hook, like fsynching a file to disk,
* before replying to a client. */
// 这个fe 是代表这个事件的总状态,首先要保证读事件,要before 于写事件,这样的好处在于我们能够将处理好的命令回滚,
// 而不至于一些同步问题导致数据无法回滚,给予客户端错误的状态,比如我们 fsyching 数据到硬盘,我们想成功之后才将数据返回给客户端
int invert = fe->mask & AE_BARRIER;
/* 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.
*
* Fire the readable event if the call sequence is not
* inverted. */
if (!invert && fe->mask & mask & AE_READABLE) {
//在这里调用我们的readder handler
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
}
/* Fire the writable event. */
if (fe->mask & mask & AE_WRITABLE) {
if (!fired || fe->wfileProc != fe->rfileProc) {
fe->wfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
}
}
/* If we have to invert the call, fire the readable event now
* after the writable one. */
if (invert) {
fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
if ((fe->mask & mask & AE_READABLE) &&
(!fired || fe->wfileProc != fe->rfileProc))
{
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
fired++;
}
}
processed++;
}
}
/* Check time events */
if (flags & AE_TIME_EVENTS)
processed += processTimeEvents(eventLoop);
return processed; /* return the number of processed file/time events */
}
从上面代码可以看到如何处理file event的过程
1.我们先调用到aeApiPoll,获取需要被处理的event个数,
2.我们传入的eventloop 里面放入了需要被处理的事件。
3.遍历eventloop里面的每一项。
4.对于可读事件调用read handler 来处理。
5.对于可写事件调用write handler 来处理。
上面还有一些回滚的处理流程,这个再后续功能模块再继续讨论。
初始化客户端连接的read handler处理流程
// 这个是服务端fd注册的读事件响应,它的作用主要用于接收到新的客户端连接,然后将它注册到epoll里面去,后面接收客户端数据的handler 就不在这边处理了。
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
char cip[NET_IP_STR_LEN];
UNUSED(el);
UNUSED(mask);
UNUSED(privdata);
while(max--) {
//跟客户端建立通道,为客户端分配一个fd.
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
serverLog(LL_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
//connCreateAcceptedSocket 主要用于初始化客户端的连接
//acceptCommonHandler 这个方法适用于接收数据的地方
acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);
}
}
//初始化一个连接,
connection *connCreateAcceptedSocket(int fd) {
connection *conn = connCreateSocket();
conn->fd = fd;
conn->state = CONN_STATE_ACCEPTING;
return conn;
}
//初始化连接
connection *connCreateSocket() {
//分配空间
connection *conn = zcalloc(sizeof(connection));
//CT_Socket 是一个结构体
conn->type = &CT_Socket;
conn->fd = -1;
return conn;
}
connection *connCreateSocket() {
connection *conn = zcalloc(sizeof(connection));
conn->type = &CT_Socket;
conn->fd = -1;
return conn;
}
//可以看到新的socket 里面分配了各种handler
ConnectionType CT_Socket = {
.ae_handler = connSocketEventHandler,
.close = connSocketClose,
.write = connSocketWrite,
.read = connSocketRead,
.accept = connSocketAccept,
.connect = connSocketConnect,
.set_write_handler = connSocketSetWriteHandler,
.set_read_handler = connSocketSetReadHandler,
.get_last_error = connSocketGetLastError,
.blocking_connect = connSocketBlockingConnect,
.sync_write = connSocketSyncWrite,
.sync_read = connSocketSyncRead,
.sync_readline = connSocketSyncReadLine,
.get_type = connSocketGetType
};
static void acceptCommonHandler(connection *conn, int flags, char *ip) {
......
//为客户端连接分配一个接收数据的结构体
// 并将新的连接放入epoll 里面
if ((c = createClient(conn)) == NULL) {
serverLog(LL_WARNING,
"Error registering fd event for the new client: %s (conn: %s)",
connGetLastError(conn),
connGetInfo(conn, conninfo, sizeof(conninfo)));
connClose(conn); /* May be already closed, just ignore errors */
return;
}
......
client *createClient(connection *conn) {
client *c = zmalloc(sizeof(client));
/* passing NULL as conn it is possible to create a non connected client.
* This is useful since all the commands needs to be executed
* in the context of a client. When commands are executed in other
* contexts (for instance a Lua script) we need a non connected client. */
if (conn) {
connNonBlock(conn);
connEnableTcpNoDelay(conn);
if (server.tcpkeepalive)
connKeepAlive(conn,server.tcpkeepalive);
//设置readhandler ,readQueryFromClient
// 就是我们上篇我们讲的io多线程读的地方了
connSetReadHandler(conn, readQueryFromClient);
connSetPrivateData(conn, c);
}
......
static inline int connSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
//这个地方将会调用到connSocketSetReadHandler,
//func 就是上面的readQueryFromClient
return conn->type->set_read_handler(conn, func);
}
//设置新的reader handler
static int connSocketSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
//
if (func == conn->read_handler) return C_OK;
//将readhandler 覆盖成readQueryFromClient
conn->read_handler = func;
if (!conn->read_handler)
aeDeleteFileEvent(server.el,conn->fd,AE_READABLE);
else
//将新的fd放入到epoll里面
//新的fd有可读事件的时候回调函数是connSocketEventHandler
if (aeCreateFileEvent(server.el,conn->fd,
AE_READABLE,conn->type->ae_handler,conn) == AE_ERR) return C_ERR;
return C_OK;
}
......
//这个地方是接收数据处理
if (connAccept(conn, clientAcceptHandler) == C_ERR) {
char conninfo[100];
if (connGetState(conn) == CONN_STATE_ERROR)
serverLog(LL_WARNING,
"Error accepting a client connection: %s (conn: %s)",
connGetLastError(conn), connGetInfo(conn, conninfo, sizeof(conninfo)));
freeClient(connGetPrivateData(conn));
return;
}
}
....
//这里会调用到CT_Socket 里面的connSocketAccept方法
static inline int connAccept(connection *conn, ConnectionCallbackFunc accept_handler) {
return conn->type->accept(conn, accept_handler);
}
static int connSocketAccept(connection *conn, ConnectionCallbackFunc accept_handler) {
int ret = C_OK;
//判断状态
if (conn->state != CONN_STATE_ACCEPTING) return C_ERR;
conn->state = CONN_STATE_CONNECTED;
connIncrRefs(conn);
//这里又会调用到clientAcceptHandler
// 而clientAcceptHandler 看作者注释主要对于一些安全模式的验证,
// 在这个环节暂时不研究
if (!callHandler(conn, accept_handler)) ret = C_ERR;
connDecrRefs(conn);
return ret;
}
以上代码总结下来主线主要就是看了以下几件事
1,当有新连接过来的时候,分配一个fd给这个客户端。
2,将这个新的fd加入到epoll体系里面,即上面的handler 只负责新的连接初始化的工作。
3,一些安全验证,还有redis额外模块加载(丰富其拓展性)等事情。
初始化新的连接后又可以回到下一轮的epoll_wait,这个时候就能从刚刚连接里面把数据给读上来了。
读客户端连接数据的handler处理流程
/**
* 可以看到下面的流程就是标准的读写流程,这边暂时来研究读的这部分
* @param el
* @param fd
* @param clientData
* @param mask
*/
static void connSocketEventHandler(struct aeEventLoop *el, int fd, void *clientData, int mask)
{
UNUSED(el);
UNUSED(fd);
.......
int call_write = (mask & AE_WRITABLE) && conn->write_handler;
int call_read = (mask & AE_READABLE) && conn->read_handler;
/* Handle normal I/O flows */
if (!invert && call_read) {
//这里就会调用到readQueryFromClient
if (!callHandler(conn, conn->read_handler)) return;
}
/* Fire the writable event. */
if (call_write) {
if (!callHandler(conn, conn->write_handler)) return;
}
/* If we have to invert the call, fire the readable event now
* after the writable one. */
if (invert && call_read) {
if (!callHandler(conn, conn->read_handler)) return;
}
.......
}
当新的fd有可读事件的时候,会从这里把数据读到client buffer
总结:
这篇文章讲述了从redis如何通过网络监听的方式到建立连接到读取数据的整个流程。能够学习到的知识点就是redis如何去使用网络初始化,如何去监听端口,又如何使用io多路复用的方式,全篇干货比较多,需要细心阅读,本篇文章最好能够对着代码,延着作者思路对照着读,文中也给出了比较详细的注释,希望能为你读源码的时候提供帮助
读源码小技巧分享
可以看到上面的文章里面其实还有一些判断和其它分支,并没有给出详细的解释,因为作者并不是redis的源码编写者,编写者本身对其有很多细节思考,但是我们不必对每个细节都去展开深入,这样会妨碍对于主线的探索,可以通过读懂其主线代码,然后再返回过去看其分支细节,这样就会有茅塞顿开的感觉。
相关推荐
- 快递查询教程,批量查询物流,一键管理快递
-
作为商家,每天需要查询许许多多的快递单号,面对不同的快递公司,有没有简单一点的物流查询方法呢?小编的回答当然是有的,下面随小编一起来试试这个新技巧。需要哪些工具?安装一个快递批量查询高手快递单号怎么快...
- 一键自动查询所有快递的物流信息 支持圆通、韵达等多家快递
-
对于各位商家来说拥有一个好的快递软件,能够有效的提高自己的工作效率,在管理快递单号的时候都需要对单号进行表格整理,那怎么样能够快速的查询所有单号信息,并自动生成表格呢?1、其实方法很简单,我们不需要一...
- 快递查询单号查询,怎么查物流到哪了
-
输入单号怎么查快递到哪里去了呢?今天小编给大家分享一个新的技巧,它支持多家快递,一次能查询多个单号物流,还可对查询到的物流进行分析、筛选以及导出,下面一起来试试。需要哪些工具?安装一个快递批量查询高手...
- 3分钟查询物流,教你一键批量查询全部物流信息
-
很多朋友在问,如何在短时间内把单号的物流信息查询出来,查询完成后筛选已签收件、筛选未签收件,今天小编就分享一款物流查询神器,感兴趣的朋友接着往下看。第一步,运行【快递批量查询高手】在主界面中点击【添...
- 快递单号查询,一次性查询全部物流信息
-
现在各种快递的查询方式,各有各的好,各有各的劣,总的来说,还是有比较方便的。今天小编就给大家分享一个新的技巧,支持多家快递,一次能查询多个单号的物流,还能对查询到的物流进行分析、筛选以及导出,下面一起...
- 快递查询工具,批量查询多个快递快递单号的物流状态、签收时间
-
最近有朋友在问,怎么快速查询单号的物流信息呢?除了官网,还有没有更简单的方法呢?小编的回答当然是有的,下面一起来看看。需要哪些工具?安装一个快递批量查询高手多个京东的快递单号怎么快速查询?进入快递批量...
- 快递查询软件,自动识别查询快递单号查询方法
-
当你拥有多个快递单号的时候,该如何快速查询物流信息?比如单号没有快递公司时,又该如何自动识别再去查询呢?不知道如何操作的宝贝们,下面随小编一起来试试。需要哪些工具?安装一个快递批量查询高手快递单号若干...
- 教你怎样查询快递查询单号并保存物流信息
-
商家发货,快递揽收后,一般会直接手动复制到官网上一个个查询物流,那么久而久之,就会觉得查询变得特别繁琐,今天小编给大家分享一个新的技巧,下面一起来试试。教程之前,我们来预览一下用快递批量查询高手...
- 简单几步骤查询所有快递物流信息
-
在高峰期订单量大的时候,可能需要一双手当十双手去查询快递物流,但是由于逐一去查询,效率极低,追踪困难。那么今天小编给大家分享一个新的技巧,一次能查询多个快递单号的物流,下面一起来学习一下,希望能给大家...
- 物流单号查询,如何查询快递信息,按最后更新时间搜索需要的单号
-
最近有很多朋友在问,如何通过快递单号查询物流信息,并按最后更新时间搜索出需要的单号呢?下面随小编一起来试试吧。需要哪些工具?安装一个快递批量查询高手快递单号若干怎么快速查询?运行【快递批量查询高手】...
- 连续保存新单号功能解析,导入单号查询并自动识别批量查快递信息
-
快递查询已经成为我们日常生活中不可或缺的一部分。然而,面对海量的快递单号,如何高效、准确地查询每一个快递的物流信息,成为了许多人头疼的问题。幸运的是,随着科技的进步,一款名为“快递批量查询高手”的软件...
- 快递查询教程,快递单号查询,筛选更新量为1的单号
-
最近有很多朋友在问,怎么快速查询快递单号的物流,并筛选出更新量为1的单号呢?今天小编给大家分享一个新方法,一起来试试吧。需要哪些工具?安装一个快递批量查询高手多个快递单号怎么快速查询?运行【快递批量查...
- 掌握批量查询快递动态的技巧,一键查找无信息记录的两种方法解析
-
在快节奏的商业环境中,高效的物流查询是确保业务顺畅运行的关键。作为快递查询达人,我深知时间的宝贵,因此,今天我将向大家介绍一款强大的工具——快递批量查询高手软件。这款软件能够帮助你批量查询快递动态,一...
- 从复杂到简单的单号查询,一键清除单号中的符号并批量查快递信息
-
在繁忙的商务与日常生活中,快递查询已成为不可或缺的一环。然而,面对海量的单号,逐一查询不仅耗时费力,还容易出错。现在,有了快递批量查询高手软件,一切变得简单明了。只需一键,即可搞定单号查询,一键处理单...
- 物流单号查询,在哪里查询快递
-
如果在快递单号多的情况,你还在一个个复制粘贴到官网上手动查询,是一件非常麻烦的事情。于是乎今天小编给大家分享一个新的技巧,下面一起来试试。需要哪些工具?安装一个快递批量查询高手快递单号怎么快速查询?...
你 发表评论:
欢迎- 一周热门
- 最近发表
- 标签列表
-
- wireshark怎么抓包 (75)
- qt sleep (64)
- cs1.6指令代码大全 (55)
- factory-method (60)
- sqlite3_bind_blob (52)
- hibernate update (63)
- c++ base64 (70)
- nc 命令 (52)
- wm_close (51)
- epollin (51)
- sqlca.sqlcode (57)
- lua ipairs (60)
- tv_usec (64)
- 命令行进入文件夹 (53)
- postgresql array (57)
- statfs函数 (57)
- .project文件 (54)
- lua require (56)
- for_each (67)
- c#工厂模式 (57)
- wxsqlite3 (66)
- dmesg -c (58)
- fopen参数 (53)
- tar -zxvf -c (55)
- 速递查询 (52)