首页 > 数据库 > redis7源码分析:redis 启动流程

redis7源码分析:redis 启动流程

时间:2023-10-02 13:22:37浏览次数:45  
标签:int handler redis7 eventLoop redis connection 源码 fd conn

1. redis 由 server.c 的main函数启动

int main(int argc, char **argv) { 
...
// 上面的部分为读取配置和启动命令参数解析,看到这一行下面为启动流程
    serverLog(LL_WARNING, "oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo");
...
	// 这里对服务进行初始化操作
    initServer();
...
	aeMain(server.el);
}

main 函数中最重要的几步

1.1 第一步就是执行 initServer

void initServer(void) {
...
    // 创建eventLoop, 这里是redis非常关键的一个结构体,在这里监听所有的事件
    server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
...
    /* 这里监听TCP端口,拿到监听的文件描述符ipfd, 监听客户端发过来的命令 */
    if (server.port != 0 &&
        listenToPort(server.port,&server.ipfd) == C_ERR) {
        /* Note: the following log text is matched by the test suite. */
        serverLog(LL_WARNING, "Failed listening on port %u (TCP), aborting.", server.port);
        exit(1);
    }
...
    // 创建定时执行的后台操作事件,定时调用serverCron 函数,
    if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
        serverPanic("Can't create event loop timers.");
        exit(1);
    }
...
    // 绑定ipfd的accept事件的执行函数 acceptTcpHandler
    // 实际是监听ipfd的可读事件,一旦可读,证明有连接进来了,立刻ceptTcpHandler
    if (createSocketAcceptHandler(&server.ipfd, acceptTcpHandler) != C_OK) {
        serverPanic("Unrecoverable error creating TCP socket accept handler.");
    }
...
    //注册eventLoop循环的执行前和执行后操作函数,类似于spring的环绕增强
    aeSetBeforeSleepProc(server.el,beforeSleep);
    aeSetAfterSleepProc(server.el,afterSleep);

}
initServer 中最重要的两步
1. 创建了eventLoop,这是redis的核心
2. 监听了TCP端口,为后续网络操作的入口(select epoll等)
3. 绑定了TCP事件的handler函数

1.2 初始化完server后,执行aeMain, ae.c是redis的核心,是一个事件驱动开发的库

void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
        aeProcessEvents(eventLoop, AE_ALL_EVENTS|
                                   AE_CALL_BEFORE_SLEEP|
                                   AE_CALL_AFTER_SLEEP);
    }
}

这里相当于是一个死循环,不停得监听事件并调用绑定的handler函数

int aeProcessEvents(aeEventLoop *eventLoop, int flags) {
...
    if (eventLoop->maxfd != -1 ||
        ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
        // 执行调用前函数
        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. */
        // 这里根据不同的操作系统执行不同的系统调用,总体就是找到可读或可写的fd
        numevents = aeApiPoll(eventLoop, tvp);

        /* 执行调用后函数. */
        if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
            eventLoop->aftersleep(eventLoop);

        for (j = 0; j < numevents; j++) {
            int fd = eventLoop->fired[j].fd;
            aeFileEvent *fe = &eventLoop->events[fd];
            int mask = eventLoop->fired[j].mask;
            int fired = 0; /* Number of events fired for current fd. */
            ...

            // aeApiPoll 中已标记好事件的类型,这里通过mask来找到可读事件
            if (!invert && fe->mask & mask & AE_READABLE) {
                // 执行绑定的可读事件的 handler函数
                fe->rfileProc(eventLoop,fd,fe->clientData,mask);
                fired++;
                fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
            }

            // aeApiPoll 中已标记好事件的类型,这里通过mask来找到可写事件
            if (fe->mask & mask & AE_WRITABLE) {
                if (!fired || fe->wfileProc != fe->rfileProc) {
                    // 执行绑定的可写事件的 handler函数
                    fe->wfileProc(eventLoop,fd,fe->clientData,mask);
                    fired++;
                }
            }
...
            processed++;
        }
    }

}

1.2 分析ae_epoll.c,这里通过系统调用epoll,找到可读可写的fd

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, numevents = 0;

    // epfd = epoll fd, 需要监听的fd都会通过epoll_ctl 提前挂到 epfd上
    // 这里等待查看可用的fd
    retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
            tvp ? (tvp->tv_sec*1000 + (tvp->tv_usec + 999)/1000) : -1);
    if (retval > 0) {
        int j;

        numevents = retval;
        // 遍历可用的fd,并通过el的filed数组进行标记,为后续执行做准备,
        // 这里仅是标记
        for (j = 0; j < numevents; j++) {
            int mask = 0;
            struct epoll_event *e = state->events+j;

            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;
        }
    } else if (retval == -1 && errno != EINTR) {
        panic("aeApiPoll: epoll_wait, %s", strerror(errno));
    }

    return numevents;
}

1.3 由于之前绑定了acceptTcpHandler, 当ipfd 可读时会调用acceptTcpHandler

acceptTcpHandler 在 networking.c中

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--) {
        // 获取accept client对应的fd,这里由于通过epoll,避免了accept的长期阻塞
        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);
		// 1. 根据client fd 创建了socket
		// 2. 执行了acceptCommonHandler
        acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);
    }
}

1.3.1 创建socket

connection.c

typedef struct ConnectionType {
    void (*ae_handler)(struct aeEventLoop *el, int fd, void *clientData, int mask);
    int (*connect)(struct connection *conn, const char *addr, int port, const char *source_addr, ConnectionCallbackFunc connect_handler);
    int (*write)(struct connection *conn, const void *data, size_t data_len);
    int (*writev)(struct connection *conn, const struct iovec *iov, int iovcnt);
    int (*read)(struct connection *conn, void *buf, size_t buf_len);
    void (*close)(struct connection *conn);
    int (*accept)(struct connection *conn, ConnectionCallbackFunc accept_handler);
    int (*set_write_handler)(struct connection *conn, ConnectionCallbackFunc handler, int barrier);
    int (*set_read_handler)(struct connection *conn, ConnectionCallbackFunc handler);
    const char *(*get_last_error)(struct connection *conn);
    int (*blocking_connect)(struct connection *conn, const char *addr, int port, long long timeout);
    ssize_t (*sync_write)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
    ssize_t (*sync_read)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
    ssize_t (*sync_readline)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
    int (*get_type)(struct connection *conn);
} ConnectionType;

ConnectionType CT_Socket = {
    .ae_handler = connSocketEventHandler,
    .close = connSocketClose,
    .write = connSocketWrite,
    .writev = connSocketWritev,
    .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
};

connection *connCreateSocket() {
    connection *conn = zcalloc(sizeof(connection));
    conn->type = &CT_Socket;
    conn->fd = -1;

    return conn;
}

// 该函数创建了一个connection 对象,绑定了fd
connection *connCreateAcceptedSocket(int fd) {
    connection *conn = connCreateSocket();
    conn->fd = fd;
    conn->state = CONN_STATE_ACCEPTING;
    return conn;
}

1.3.2 执行 acceptCommonHandler

static void acceptCommonHandler(connection *conn, int flags, char *ip) {
...
    // 最关键的一步,创建client
    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;
    }

...
}

1.3.3 创建出client,与客户端一一对应

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) {
        connEnableTcpNoDelay(conn);
        if (server.tcpkeepalive)
            connKeepAlive(conn,server.tcpkeepalive);
        // 关键一步,在这里绑定读取事件 readQueryFromClient
        connSetReadHandler(conn, readQueryFromClient);
        // 将connection 与 client 绑定
        connSetPrivateData(conn, c);
    }
    ...

}

1.3.4 读事件绑定,这里是client 发命令到 server 读取的入口

connection.h

/* Register a read handler, to be called when the connection is readable.
 * If NULL, the existing handler is removed.
 */
static inline int connSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
    return conn->type->set_read_handler(conn, func);
}
由connection 构造时的逻辑可知, 这里的type 是 CT_Socket,那么 set_read_handler 是 connSocketSetReadHandler
static int connSocketSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
    if (func == conn->read_handler) return C_OK;

    conn->read_handler = func;
    if (!conn->read_handler)
        aeDeleteFileEvent(server.el,conn->fd,AE_READABLE);
    else
        // 这里注册了一个可读的事件,用来回调read_handler,即 readQueryFromClient
        // 这里的fd 是accept 得到的 client fd
        if (aeCreateFileEvent(server.el,conn->fd,
                    AE_READABLE,conn->type->ae_handler,conn) == AE_ERR) return C_ERR;
    return C_OK;
}
到这里注册客户端发送来命令的可读事件就注册完成了,启动逻辑就完成了

到这里还是在一条线程中执行的所有指令,那么 redis 是否就是一个线程搞完所有事情呢

其实不是,redis 有两种模式,一种是单线程模式,一种是多线程模式

将在后续的文章中继续分析

标签:int,handler,redis7,eventLoop,redis,connection,源码,fd,conn
From: https://www.cnblogs.com/gradyblog/p/17739866.html

相关文章

  • Python爬虫源码,Behance 作品图片及内容 selenium 采集爬虫
    前面有分享过requests采集Behance作品信息的爬虫,这篇带来另一个版本供参考,使用的是无头浏览器selenium采集,主要的不同方式是使用selenium驱动浏览器获取到页面源码,后面获取信息的话与前篇一致。Python爬虫源码,Behance作品图片及内容采集爬虫附工具脚本!理论上,几乎所有的页面内......
  • redis 源码分析:Jedis 哨兵模式连接原理
    1.可以从单元测试开始入手查看类JedisSentinelPoolprivatestaticfinalStringMASTER_NAME="mymaster";protectedstaticfinalHostAndPortsentinel1=HostAndPorts.getSentinelServers().get(1);protectedstaticfinalHostAndPortsentinel2=HostAndPorts......
  • Redis数据结构
    本文大部分知识整理自网上,在正文结束后都会附上参考地址。如果想要深入或者详细学习可以通过文末链接跳转学习。前言本文主要介绍关于Redis的五种基本数据结构的底层实现原理,然后来分析我们常用的使用场景。先简单回顾一下知识点。Redis是一个开源(BSD许可)的,内存中的数据结......
  • token+redis的简单使用方式
    以用户登录为例,讲解token+redis的使用方式,环境是vue和springboot。一、用户登录时序图二、前端代码分析1、前端使用vuestore保存token2、在每次发起请求时进行响应拦截,从vuestore取出token,放在每次请求的请求头上三、后端代码分析1、在控制层接收账号,密码,调用服务层代......
  • Redis 常见面试题总结
    什么是Redis?Redis(RemoteDictionaryServer)是一个开源的使用ANSIC语言编写、遵守BSD协议、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库,并提供多种语言的API的非关系型数据库。传统数据库遵循ACID规则。而Nosql(非关系型数据库)一般为分布式而分布式一......
  • Redis实现分布式锁
    一、分布式锁参考资料:www.cnblogs.com/wangyingshu…很多场景中,需要使用分布式事务、分布式锁等技术来保证数据最终一致性。有的时候,我们需要保证某一方法同一时刻只能被一个线程执行。在单机(单进程)环境中,JAVA提供了很多并发相关API,但在多机(多进程)环境中就无能为力了。对于分......
  • springboot 与 Redis整合
    SpringBoot操作数据:Spring-datajpajdbcmongodbredis!SpringData也是和SpringBoot齐名的项目!说明:在SpringBoot2.X之后,原来使用的jedis被替换成了lettucejedis:采用的直连,多个线程操作的话,是不安全的,如果想要避免不安全的,使用jedispool连接池,更新BIO模式lettuce:采用ne......
  • Redis哨兵集群原理
    单节点Redis的并发能力是有上限的,要进一步提高Redis的并发能力,就需要搭建主从集群,实现读写分离主节点:可以对Redis实现读写操作从节点: 只可以对Redis实现读操作但是,当master节点宕机后,我们就不能写数据到Redis,所以需要搭建一个三节点形成的Sentinel集群,来监管之前的Redis主从集......
  • diskqueue的数据定义,运转核心ioloop()源码详解
     nsq中diskqueue是nsq消息持久化的核心,内容较多,一共分为多篇1.diskqueue是什么,为什么需要它,整体架构图,对外接口2.diskqueue的元数据文件,数据文件,启动入口,元数据文件读写及保存3.diskqueue的数据定义,运转核心ioloop()源码详解4. diskqueue怎么写入消息,怎么对外发送消息前面一篇......
  • Go每日一库之161:grm(Redis Web管理工具)
    GRM是基于go+vue的web版redis管理工具,部署简单便捷,支持SSH连接,用户校验,操作日志、命令行模式、LUA脚本执行等功能。介绍基于go+vue的web版redis管理工具【Webredismanagementtoolbasedongolangandvue】功能清单管理连接(直连和SSH)、切换DB支持string/lis......