怎么做盗版电影网站吗,企业网站的功能可分为前台和后台两个部分,杭州品牌网站设计,做网站时怎么选择数据库类型前言#xff1a;做码农这么多年#xff0c;我也读过很多开源软件或者框架的源码#xff0c;在我看来#xff0c;Redis是我看过写得最优美、最像一件艺术品的软件#xff0c;正如Redis之父自己说的那样#xff0c;他宁愿以一个糟糕的艺术家身份而不是一名好程序员被别人记… 前言做码农这么多年我也读过很多开源软件或者框架的源码在我看来Redis是我看过写得最优美、最像一件艺术品的软件正如Redis之父自己说的那样他宁愿以一个糟糕的艺术家身份而不是一名好程序员被别人记住我认为他不仅做到了而且还是一个非常高超的艺术家。记得当年看源码时我不禁发出感叹原来用C语言也可以写出如此美妙、优雅的代码来 Redis经过十多年的发展几乎是所有公司必不可少的中间件了要么用其做缓存要么做数据统计或者分布式锁有些电商还会用Redis做购物车突然让我想起来那是在2019年前东家当时我们用的是codis搭的Redis集群Redis集群挂了导致大促时大家都操作不了购物车这家把老板急的现在还印象深刻哈哈。。Redis之所以如此受到欢迎主要还是得益于其丰富的数据结构和较高的性能。Redis官方给出了单机的Redis性能在忽略带宽影响的前提下Redis的单机QPS可达到10万QPS。下图是使用Redis benchmark做的性能测试 Redis的卓越性能主要原于以下几个方面
使用内存进行数据读写IO多路复用复用思想提高请求处理能力后台线程处理耗时任务异步线程不阻塞主线程IO多线程进一步提升网络请求处理性能
1、使用内存 多数情况下我们编写的计算机任务都是IO密集型的80%甚至更多的时间都是在处理各种IO磁盘IO网络IO等等Redis为了提升读写性能让主线程的所有读写完全是在内存中进行从而不需要数据磁盘和内核缓冲区之间的拷贝我认为这是其性能较好的最重要原因。 虽然Redis使用内存存储但和Memcached不同的是Redis是支持持久化的即内存中的数据是可以持久化到磁盘文件中从而可保证在断电重启后恢复数据以及支持主从复制。Redis的持久化包含RDB以及AOF持久化两种。 RDB类似Mysql基于Row模式的binlog即存储的是数据本身是一种数据快照RDB在全量复制以及断电恢复中非常有用由于都是数据所以恢复较快。Redis生成RDB文件的过程并不是在主进程中完成其会fork出一个子进程来完成RDB文件生成并不会阻塞主进程此外其还充分利用了写时复制COW的机制fork的子进程依然会和父进程共享物理内存因此不会影响主进程处理命令的性能这里不考虑内存等因素的影响实现命令是bgsave。 AOFAppend only file文件记录格式类似于Mysql的基于statment记录的binlong即记录的是执行命令当然其只会记录写命令读命令不会记录。当执行完写命令后会直接返回客户端随后主线程还是会将命令写入到AOF文件中这个过程不会影响当前命令但会影响下一个命令因为其是在主进程中进行的。写完AOF之后实际上此时数据是在PageCache中我在之前的Linux零拷贝技术浅谈中介绍过随后Redis可以主动地将数据刷盘比如每秒刷一次(fsync这个也是后台线程处理的)也可以同步刷但会影响性能也可以直接不管由操作系统刷盘。 另外一个需要提的是AOF的重写AOF如果不断地追加命令会导致文件过大因此为了缩小AOF的文件会进行重写重写的目的是将命令进行整合比如对同一个Key的写命令整合成一个从而缩小文件大小。重写的操作也是通过子进程完成的并不会阻塞主进程。
2、Redis的I/O多路复用 Redis采用单线程IO多路复用来实现高并发即著名的Reactor单线程模式默认会使用epoll机制通过回调函数、红黑树、mmap机制等各种骚操作提升请求处理能力。其中epoll_create用来创建epoll对象epoll_ctl用于将就绪事件添加到epoll对象中epoll_wait会检查对象中是否存在就绪事件如果有就会把数据拷贝从内核拷贝到用户空间。 具体的关于多路复用和Reactor模式的介绍可以看我之前的文章IO多路复用介绍 Java的NIO及Netty 。 这里可能你有疑问为啥Redis另辟蹊径只使用单线程的IO多路复用这也无法充分发挥多核的优势啊无法使用是多核是真的但Redis由于本身使用内存读写数据因此多数情况下瓶颈并不在CPU上且使用单线程读写数据并不需要考虑线程安全性不必加锁一定程度简化了Redis的复杂度这也算是作者做的一种权衡。可是聪明的你可能还会继续持怀疑态度如果执行命令耗时较长不会阻塞吗 网络请求过大时网络IO还是会阻塞啊 是的对待这两个问题Redis在后续版本中都给了解决方案对应的就是第三和第四小节。 下面是我在网上发现一个清晰得描述Redis的IO多路复用实现机制得示意图来自飞哥开发内功修炼 Redis服务端的处理流程
Redis 服务启动调用main函数initServer主要进行主线程事件循环aeCreateEventLoop注册 acceptTcpHandler 函数等待新连接到来。看到eventLoop大家应该熟悉在Netty中你也会看到的这是IO多路复用的核心客户端和服务端建立Socket连接Redis Server多路复用接收请求并调用acceptTcpHandler 函数最终调用 readQueryFromClient 命令读取并解析客户端连接客户端发送请求触读就绪事件主线程调用 readQueryFromClient 读取客户端命令并写入querybuf 读入缓冲区接着调用 processInputBuffer随后调用 processCommand 执行命令执行命令后将响应数据写入到对应客户端的写缓冲区固定大小 16KB一般来说可以缓冲足够多的响应数据但是如果客户端在时间窗口内需要响应的数据非常大那么则会自动切换到 client-reply 链表上去最后把 client 添加进一个 LIFO 队列 clients_pending_write在事件文件或定时循环中主线程执行 beforeSleep -- handleClientsWithPendingWrites遍历 clients_pending_write 队列调用 writeToClient 把 客户端写出缓冲区里的数据发送到客户端如果写出缓冲区还有数据遗留则注册 sendReplyToClient 命令回复处理器到该连接的写就绪事件等待客户端可写时在事件循环中再继续回写剩余的响应数据。 接下来具体看看Redis的实现入口是src/server.c的main函数。 1、这个函数会进行环境设置、初始化参数、数据恢复等等一系列操作。
int main(int argc, char **argv) {initServer();aeMain(server.el);aeDeleteEventLoop(server.el);return 0;
}initServer用于创建epoll注册定时和文件事件这是Redis的最重要两类事件。
文件事件file eventRedis客户端通过socket与Redis服务器连接而文件事件就是服务器对套接字操作的抽象。例如客户端发了一个GET命令请求对于Redis服务器来说就是一个文件事件。Redis的协议本身发送的是文本。时间事件time event服务器定时或周期性执行的事件。例如定期执行RDB持久化。 //创建epoll对象server.el aeCreateEventLoop(server.maxclientsCONFIG_FDSET_INCR);epoll对象创建过程
aeEventLoop *aeCreateEventLoop(int setsize) {aeEventLoop *eventLoop;int i;if (aeApiCreate(eventLoop) -1) goto err;return eventLoop;
}重要的就是这句aeApiCreate其用于创建实际的epoll对象。
static int aeApiCreate(aeEventLoop *eventLoop) {aeApiState *state zmalloc(sizeof(aeApiState));//露出真面目了state-epfd epoll_create(1024); eventLoop-apidata state;return 0;
}2、随后开始注册回调行数initServer中注册的函数acceptTcpHandler当有新连接到来时该函数会被执行。 //注册定时事件用于执行后台一些业务如定时清理if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) AE_ERR) {serverPanic(Cant create event loop timers.);exit(1);}//创建文件事件处理器用于处理tcp连接。for (j 0; j server.ipfd_count; j) {if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,acceptTcpHandler,NULL) AE_ERR){serverPanic(Unrecoverable error creating server.ipfd file event.);}}再看一下aeCreateFileEvent函数
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,aeFileProc *proc, void *clientData)
{//关键的两句aeFileEvent *fe eventLoop-events[fd];if (aeApiAddEvent(eventLoop, fd, mask) -1)return AE_ERR;fe-mask | mask;if (mask AE_READABLE) fe-rfileProc proc;if (mask AE_WRITABLE) fe-wfileProc proc;fe-clientData clientData;return AE_OK;
}aeApiAddEvent会将文件描述符添加到epoll队列中实际上调用的就是epoll_ctl。至此Server初始化基本完成。
接着执行aeMain过程进入死循环
void aeMain(aeEventLoop *eventLoop) {eventLoop-stop 0;while (!eventLoop-stop) {aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_BEFORE_SLEEP|AE_CALL_AFTER_SLEEP);}
}aeProcessEvents调用epoll_wait阻塞等待事件就绪。
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{//这里是调用epoll_wait阻塞numevents aeApiPoll(eventLoop, tvp);for (j 0; j numevents; j) {// 从已就绪队列中获取事件aeFileEvent *fe eventLoop-events[eventLoop-fired[j].fd];//如果是读事件并且有读回调函数fe-rfileProc()//如果是写事件并且有写回调函数fe-wfileProc()}}static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {// 等待事件aeApiState *state eventLoop-apidata;epoll_wait(state-epfd,state-events,eventLoop-setsize,tvp ? (tvp-tv_sec*1000 tvp-tv_usec/1000) : -1);...
}这里具体实现机制取决于系统支持哪些如果不支持epoll默认会使用select。 假如现在有新连接到来了此时会调用已注册的acceptTCPHandler函数我们看下其具体处理流程。
acceptTCPHandler
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {while(max--) {//调用accept接收连接cfd anetTcpAccept(server.neterr, fd, cip, sizeof(cip), cport);acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);}
}2、然后执行acceptCommonHandler
static void acceptCommonHandler(connection *conn, int flags, char *ip) {//创建Redis客户端if ((c createClient(conn)) NULL) {connClose(conn); /* May be already closed, just ignore errors */return;}
}3、创建连接客户端
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);//画重点connSetReadHandler(conn, readQueryFromClient);connSetPrivateData(conn, c);}
}上面首先创建了一个读事件。
connSetReadHandler(conn, readQueryFromClient);4、从客户端读取数据
readQueryFromClient该函数负责读取客户端命令。
5、调用processCommand执行命令执行具体操作
int processCommandAndResetClient(lient *c) {if (processCommand(c) C_OK) {commandProcessed(c);}
}注意Redis处理之后并不是直接将数据返给客户端而是先加入到写任务队列在每次循环首先进行数据发送。 3、后台线程 上面提到了 由于Redis的命令是单线程的那如果某个命令出现了阻塞就会影响到所有命令的执行。就拿我们公司去年发起的几个事故举例有几个系统创建了big key且对big key进行批量操作这直接导致请求量较高时Redis出现了严重的阻塞。这也是为什么使用Redis时我们要禁止耗时操作包括执行keys flush执行批量操作创建big key等。 为了解决Redis内置命令的一些耗时操作影响主线程自Redis4.0之后Redis增加了异步线程的支持使得一些比较耗时的任务可以在后台异步线程执行不必再阻塞主线程。之前del和flush等都会阻塞主线程现在的ulinkflushal async flushdb async等操作都不会再阻塞。 异步线程是通过Redis的bio实现即Background I/O不是我们大家经常说的阻塞IO啊哈哈。
Redis在启动时在后台会初始化三个后台线程。
void InitServerLast() {bioInit();initThreadedIO();set_jemalloc_bg_thread(server.jemalloc_bg_thread);server.initial_memory_usage zmalloc_used_memory();
}bioInit就是具体启动后台线程过程。启动的线程主要包括
#define BIO_CLOSE_FILE 0 /* Deferred close(2) syscall. */
#define BIO_AOF_FSYNC 1 /* Deferred AOF fsync. */
#define BIO_LAZY_FREE 2 /* Deferred objects freeing. */即用来关闭文件描述符、AOF持久化以及惰性删除。这三个线程是完全独立的互不干涉。每个线程都会有一个工作队列用于生产和消费任务。图片来源小林coding for (j 0; j BIO_NUM_OPS; j) {void *arg (void*)(unsigned long) j;if (pthread_create(thread,attr,bioProcessBackgroundJobs,arg) ! 0) {serverLog(LL_WARNING,Fatal: Cant initialize Background Jobs.);exit(1);}bio_threads[j] thread;}主线程负责把相关任务添加到对应线程的队列中在添加和移除队列中都会加锁防止并发问题。比如惰性删除的过程即我们使用UNLINKflushDB,flushall等命令时。 //添加任务到对应线程队列中添加过程会加锁
void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3) {struct bio_job *job zmalloc(sizeof(*job));//加锁pthread_mutex_lock(bio_mutex[type]);//将任务加到队列尾部listAddNodeTail(bio_jobs[type],job);pthread_mutex_unlock(bio_mutex[type]);
}接下来就是后台线程会从对应队列中取出任务执行
void *bioProcessBackgroundJobs(void *arg) {struct bio_job *job;unsigned long type (unsigned long) arg;sigset_t sigset;switch (type) {case BIO_CLOSE_FILE:redis_set_thread_title(bio_close_file);break;case BIO_AOF_FSYNC:redis_set_thread_title(bio_aof_fsync);break;case BIO_LAZY_FREE:redis_set_thread_title(bio_lazy_free);break;}4、IO多线程 上面已提到Redis主要采用单线程IO多路复用实现高并发后来为了处理耗时比较长的任务Redis4.0引入了BackGround I/O线程。本身Redis的瓶颈并不是在于CPU而是内存和网络IO。在一定程度上通过扩容即可。但是业务量不断扩大时网络IO的瓶颈就体现出来了。这个时候使用多线程处理还是挺香的可以充分发挥多核的优势因此Redis6.0就引入了多线程。 不过这里强调一点Redis引入了多线程仅仅是用来网络读写Redis命令的执行还是通过主线程顺序执行这主要是为了减少Redis操作的复杂度等方面。 上面在说启动Background IO时说到了InitServerLast里面有个initThreadedIO这个就是初始化线程IO的过程。
初始化线程IO的实现
void initThreadedIO(void) {server.io_threads_active 0; /* We start with threads not active. *//* Spawn and initialize the I/O threads. */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();//第一个是主线程创建完continueif (i 0) continue; /* Thread 0 is the main thread. *//* Things we do only for the additional threads. *///worker线程注册回调函数。pthread_t tid;pthread_mutex_init(io_threads_mutex[i],NULL);setIOPendingCount(i, 0);pthread_mutex_lock(io_threads_mutex[i]); /* Thread will be stopped. *///创建线程并注册回调函数IOThreadMainif (pthread_create(tid,NULL,IOThreadMain,(void*)(long)i) ! 0) {serverLog(LL_WARNING,Fatal: Cant initialize IO thread.);exit(1);}io_threads[i] tid;}
}看上面代码第一个是主线程其他的都是Worker Thread。
一个图形可以反映上面的实现 从上面示意图可以看出来IO处理已经由直线的单线程IO多路复用改成多线程的IO多路复用处理在网络IO处理中充分发挥多核的优势。 接下来具体看一下他是怎么实现多线程处理的。其实请求处理流程还是和上面的一致只是在readQueryFromClient有所不同。
void readQueryFromClient(connection *conn) {client *c connGetPrivateData(conn);int nread, readlen;size_t qblen;/* Check if we want to read from the client later when exiting from* the event loop. This is the case if threaded I/O is enabled. */if (postponeClientRead(c)) return;...
}如果是开启了线程IOPostoneClientRead会把事件加入到队列中待主线程分配给工作线程执行。
/* Return 1 if we want to handle the client read later using threaded I/O.* This is called by the readable handler of the event loop.* As a side effect of calling this function the client is put in the* pending read clients and flagged as such. */
int postponeClientRead(client *c) {if (server.io_threads_active server.io_threads_do_reads !ProcessingEventsWhileBlocked !(c-flags (CLIENT_MASTER|CLIENT_SLAVE|CLIENT_PENDING_READ))){c-flags | CLIENT_PENDING_READ;listAddNodeHead(server.clients_pending_read,c);return 1;} else {return 0;}
}接着在beforeSleep中会调用处理函数多线程处理read操作。注意下面的函数注解主线程也会处理一部分网络IO同时IO线程也会并发处理主线程会一直等到所有IO线程执行完。
int handleClientsWithPendingReadsUsingThreads(void) {if (!server.io_threads_active || !server.io_threads_do_reads) return 0;int processed listLength(server.clients_pending_read);if (processed 0) return 0;if (tio_debug) printf(%d TOTAL READ pending clients\n, processed);/* Distribute the clients across N different lists. */listIter li;listNode *ln;listRewind(server.clients_pending_read,li);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;}/* Give the start condition to the waiting threads, by setting the* start condition atomic var. */io_threads_op IO_THREADS_OP_READ;for (int j 1; j server.io_threads_num; j) {int count listLength(io_threads_list[j]);setIOPendingCount(j, count);}/* Also use the main thread to process a slice of clients. */listRewind(io_threads_list[0],li);while((ln listNext(li))) {client *c listNodeValue(ln);readQueryFromClient(c-conn);}listEmpty(io_threads_list[0]);//阻塞等待所有IO线程执行完毕while(1) {unsigned long pending 0;for (int j 1; j server.io_threads_num; j)pending getIOPendingCount(j);if (pending 0) break;}if (tio_debug) printf(I/O READ All threads finshed\n);//开始执行命令while(listLength(server.clients_pending_read)) {ln listFirst(server.clients_pending_read);client *c listNodeValue(ln);c-flags ~CLIENT_PENDING_READ;listDelNode(server.clients_pending_read,ln);if (c-flags CLIENT_PENDING_COMMAND) {c-flags ~CLIENT_PENDING_COMMAND;if (processCommandAndResetClient(c) C_ERR) {/* If the client is no longer valid, we avoid* processing the client later. So we just go* to the next. */continue;}}processInputBuffer(c);}return processed;
}主线程阻塞等待所有的工作线程都完成之后开始串行执行命令随后IO线程可以并行将数据发送到写任务队列。
下面是work线程的主要处理逻辑
void *IOThreadMain(void *myid) {/* The ID is the thread number (from 0 to server.iothreads_num-1), and is* used by the thread to just manipulate a single sub-array of clients. */long id (unsigned long)myid;char thdname[16];snprintf(thdname, sizeof(thdname), io_thd_%ld, id);redis_set_thread_title(thdname);redisSetCpuAffinity(server.server_cpulist);makeThreadKillable();while(1) {/* Wait for start */for (int j 0; j 1000000; j) {if (getIOPendingCount(id) ! 0) break;}/* Give the main thread a chance to stop this thread. */if (getIOPendingCount(id) 0) {pthread_mutex_lock(io_threads_mutex[id]);pthread_mutex_unlock(io_threads_mutex[id]);continue;}serverAssert(getIOPendingCount(id) ! 0);if (tio_debug) printf([%ld] %d to handle\n, id, (int)listLength(io_threads_list[id]));/* Process: note that the main thread will never touch our list* before we drop the pending count to 0. */listIter li;listNode *ln;listRewind(io_threads_list[id],li);
//处理读写while((ln listNext(li))) {client *c listNodeValue(ln);//如果是写执行writeToClientif (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]);setIOPendingCount(id, 0);if (tio_debug) printf([%ld] Done\n, id);}
}从上面可以看到IO线程要么同时读要么同时写不可同时包含读和写两部分。
通过ITNEXT平台的测试报告中可以看到在相同机器配置下加入多线程后Redis可支持的最大QPS达到 20W/s。 Redis这部分和Memcached的思想有些类似Memcached也是通过主线程IO多路复用接受连接并通过一定算法分配到工作线程。但是最大的不同是Redis的多线程只是为了处理网络读写不负责处理具体业务逻辑命令还是主线程顺序执行的。然而Memcached是Master线程把连接分配给Worker之后Worker线程就负责把处理后续的所有请求完全是多线程执行。Redis要想做到这一点需要做的工作还有很多尤其是线程安全方面。据说或许要逐步加Key-level的锁肯定是不断地完善的。
总结 本文主要介绍了redis的高性能解决方案包括内存、IO多路复用、后台线程以及IO多线程。 不知道后续的Redis发展会是什么样会不会真的在命令执行中引入多线程如果引入的话这对于Redis来说是一次大的改动。我认为很长一段时间不会这么做这个我们拭目以待吧。 我们看Redis7里面主要还是对数据结构、持久化、命令等方面做的优化和调整并没有特别大的改动。自从2020年Redis之父离开了Redis项目后我感觉Redis的发展路线会有所改变就像红楼梦的后四十回是高鹗写的你不能说不好只能说可能和曹雪芹最开始的整体规划以及结局是不同的。
