基于C语言实现内存型数据库(kv存储)

基于C语言实现内存型数据库(kv存储)


  
   

• 源代码仓库见Github：kv-store仓库
  
• 共同我的讲解视频更佳：【C语言项目笔记】基于C语言实现内存型数据库(kv存储)。
  
• 参考视频：“零声教育”的“linux基础架构-Kv存储”。
  
• 其他源码：协程。
  

  

---

1. 项目背景

1.1 Redis先容

  本项目主要想仿照Redis的交互方式，实现一个基本的“内存型数据库”，以是起首来先容一下Redis。随着互联网的遍及，只要是上网的APP基本上都需要和相应的服务器请求数据，通常来说，这些数据被服务器保存在“磁盘”上的文件中，称之为“磁盘型数据库”。但是面对海量用户时(比如秒杀运动)，磁盘IO的读写速率不够快从而导致用户体验下降，并且服务器数据库的压力也非常大。鉴于很多请求只是读取数据，这就启发我们将一些热门数据存放在内存中，以便快速响应请求、并且减轻磁盘的读写压力。
当然，上述只是一个初步的想法，后续如何清算内存数据、分布式存储等可以参考B站的科普视频，讲的非常简洁易懂：
   

• 【趣话Redis第一弹】我是Redis，MySQL大哥被我坑惨了！—“缓存穿透、缓存击穿、缓存雪崩”、“定时删除、惰性删除、内存淘汰”
  
• 【趣话Redis第二弹】Redis数据持久化AOF和RDB原理一次搞懂！—“RDB+AOF”
  
• 【趣话Redis第三弹】Redis的高可用是怎么实现的？哨兵是什么原理？—“主观下线、客观下线”、“哨兵选举”、“故障转移”
  
• 趣话Redis：Redis集群是如何工作的？—“哈希桶”、“集群工作+主从复制”
  

  下面是一些典范的口试题：
   

• 为什么要利用Redis？
  

     

• 从高并发上来说：直接操纵缓存能够承受的请求是远远大于直接访问数据库的，以是我们可以考虑把数据库中的部分数据转移到缓存中去。这样用户的一部分请求会直接到缓存，而不用颠末数据库。
  
• 从高性能上来说：用户第一次访问数据库中的某些数据，因为是从硬盘上读取的，以是这个过程会比力慢。将该用户访问的数据存在缓存中，下一次再访问这些数据的时候就可以直接从缓存中获取了。操纵缓存就是直接操纵内存，以是速度相当快。
  

   

• 为什么要利用Redis而不是其他的，比方Java自带的map或者guava？
  

       缓存分为本地缓存和分布式缓存，像map或者guava就是本地缓存。本地缓存最主要的特点是轻量以及快速，生命周期随着jvm的烧毁而结束。在多实例的情况下，每个实例都需要各自保存一份缓存，缓存不具有一致性。redis或memcached之类的称为分布式缓存，在多实例的情况下，各实例共用一份缓存数据，缓存具有一致性。
   

• Redis应用场景有哪些?
  

     

• 缓存热门数据，缓解数据库的压力。
  
• 利用Redis原子性的自增操纵，可以实现计数器的功能。比如统计用户点赞数、用户访问数等。
  
• 分布式锁。在分布式场景下，无法利用单机情况下的锁来对多个节点上的进程进行同步。可以利用Redis自带的SETNX命令实现分布式锁，除此之外，还可以利用官方提供的RedLock分布式锁实现。
  
• 简单的消息队列。可以利用Redis自身的发布/订阅模式或者List来实现简单的消息队列，实现异步操纵。
  
• 限速器。可用于限定某个用户访问某个接口的频率，比如秒杀场景用于防止用户快速点击带来不须要的压力。
  
• 好友关系。利用集合的一些命令，比如交集、并集、差集等，实现共同好友、共同爱好之类的功能。
  

   

• 为什么Redis这么快？
  

     

• Redis是基于内存进行数据操纵的Redis利用内存存储，没有磁盘IO上的开销，数据存在内存中，读写速度快。
  
• 接纳IO多路复用技能。Redis利用单线程来轮询形貌符，将数据库的操纵都转换成了变乱，不在网络I/O上浪费过多的时间。
  
• 高效的数据布局。Redis每种数据类型底层都做了优化，目的就是为了追求更快的速度。
  

   

• 参考视频：为什么要利用Redis？、Redis的应用场景有哪些？、Redis，好快！
  

  1.2 项目预期及基本架构

         图1 项目框架    于是我们现在就来实现这个“内存型数据库”，本项目利用C语言，默认键值对key-value都是char*类型。如上图所示，我们渴望“客户端”可以和“服务端”通讯，发送相应的指令并得到相应的信息。比如“客户端”插入一个新的键值对“(name: humu)”，那么就发送“SET name humu”；“服务端”接收到这个数据包后，执行相应的操纵，再返回“OK”给“客户端”。鉴于kv存储需要强查找的数据布局，我们可以利用rbtree、btree、b+tree、hash、dhash、array(数据量不多，比如http头)、skiplist、list(性能低不考虑)。终极，下表列出了我们要实现的所有数据布局及其对应的指令：
   表1 kv存储协议对应的数据布局及命令    操纵/
数据布局插入查找删除计数存在arraySET key valueGET keyDELETE keyCOUNTEXIST keyrbtreeRSET key valueRGET keyRDELETE keyRCOUNTREXIST keybtreeBSET key valueBGET keyBDELETE keyBCOUNTBEXIST keyhashHSET key valueHGET keyHDELETE keyHCOUNTHEXIST keydhashDHSET key valueDHGET keyDHDELETE keyDHCOUNTDHEXIST keyskiplistZSET key valueZGET keyZDELETE keyZCOUNTZEXIST key备注返回OK/Fail，
表示插入键值对是否成功返回对应的value返回OK/Fail，
删除对应的键值对返回当前数据布局中
存储的键值对数量返回True/False，
判断是否存在对应的键值对    进一步，由于网络编程中的“Hello,World!程序”就是实现一个echo，收到什么数据就原封不动的发送回去。以是我们渴望在此基础上，将kv存储写成一个独立的进程，和网络收发相干代码隔脱离，进而提升代码的可维护性。另外在网络协议的选择中，由于我们的键值对设置通常较短只有十几个字符（比如set key value），而http协议的协议头就有几十个字符，有效数据占比太低；udp协议只能在底层网卡确认对方收到，但没法在应用层确认，以是不可控；于是我们网络通信协议选择tcp。于是对于“服务端”，我们就可以有如下的架构设计：
   

    图2 “服务端”程序架构   

• 网络层：负责收发数据。本项目中都是“字符串”。
  
• 协议层：将“网络层”传输过来的字符串进行拆解，若为无效指令直接返回相应的提示信息；若为有效指令则通报给“引擎层”进行进一步的处理，根据“引擎层”的处理结果给出相应的返回信息。
  
• 引擎层：分为6种存储引擎，每种存储引擎都可以进行具体的增、删、查等操纵，也就是实现上表给出的5种命令。
  
• 存储层：注意“内存型数据库”的数据在内存中，但若后续需要“持久化”也会将数据备份到磁盘中。
  

  2. 服务端原理及代码框架

2.1 网络数据回环的实现

  在利用原生的socket库函数进行网络通信时，会一直阻塞等待客户端的连接/通信请求，这个线程就做不了其他的事情，非常浪费资源。于是“reactor模式”应运而生，也被称为“基于变乱驱动”，核心点在于：注册变乱、监听变乱、处理变乱。也就是说，线程找了一个“秘书”专门负责去监听网络端口是否有“网络通信”的发生，线程就可以去做其他的事情；等到线程想处理“网络通信”的时候一起全部通知给该线程，然后这个“秘书”继续监听。显然，有这样一个“秘书”存在，可以将“网络通信”、“业务处理”分隔开，一个线程可以同时处理多个客户端的请求/通信，也就实现了“IO多路复用一个线程”。下面是常见的三种reactor模式：
   

• reactor单线程模型：只分配一个线程。显然若线程的“业务处理”时间过长，会导致“秘书”积压的变乱过多，甚至可能会丢弃一些变乱。本模型不适合计算密集型场景，只适合业务处理非常快的场景（本项目就是业务处理非常快）。
  
• reactor多线程模型：分配一个主线程和若干子线程。主线程只负责处理“网络通信”，“业务处理”则交给子线程处理。本模式的好处是可以充实利用多核CPU性能，但是带来了线程安全的问题。并且只有一个线程响应“网络通信”，在瞬时高并发的场景下容易成为性能瓶颈。
  
• 主从reactor多线程模型：在上述多线程模型的基础上，再额外开辟出新的子线程专门负责“与客户端通信”，主线程则只负责“连接请求”。
  

  参考B站视频：【Java口试】先容一下Reactor模式
  下面利用epoll作为“秘书”，接纳“reactor单线程模型”，完成网路数据回环(echo)，也就是“服务端”程序框架的“网络层”：
main.c–共356行

1. /*
2. * zv开头的变量是zvnet异步网络库（epoll）。
3. * kv开头的变量是kv存储协议解析。
4. */
5. #include<stdio.h>
6. #include<stdlib.h>
7. #include<string.h>
9. #include<errno.h>
10. #include<unistd.h>
11. #include<sys/socket.h>
12. #include<netinet/in.h>
13. #include<fcntl.h>
14. #include<sys/epoll.h>
16. #include"kvstore.h"
18. /*-------------------------------------------------------*/
19. /*-----------------------异步网路库-----------------------*/
20. /*-------------------------------------------------------*/
21. /*-----------------------函数声明-------------------------*/
22. #define max_buffer_len      1024    // 读写buffer长度
23. #define epoll_events_size   1024    // epoll就绪集合大小
24. #define connblock_size      1024    // 单个连接块存储的连接数量
25. #define listen_port_count   1       // 监听端口总数
27. // 有如下参数列表和返回之类型的函数都称为 CALLBACK
28. // 回调函数，方便在特定epoll事件发生时执行相应的操作
29. typedef int (*ZV_CALLBACK)(int fd, int events, void *arg);
30. // 回调函数：建立连接
31. int accept_cb(int fd, int event, void* arg);
32. // 回调函数：接收数据
33. int recv_cb(int clientfd, int event, void* arg);
34. // 回调函数：发送数据
35. int send_cb(int clientfd, int event, void* arg);
37. // 单个连接
38. typedef struct zv_connect_s{
39.     // 本连接的客户端fd
40.     int fd;
41.     // 本连接的读写buffer
42.     char rbuffer[max_buffer_len];
43.     size_t rcount;  // 读buffer的实际大小
44.     char wbuffer[max_buffer_len];
45.     size_t wcount;  // 写buffer的实际大小
46.     size_t next_len;  // 下一次读数据长度（读取多个包会用到）
47.     // 本连接的回调函数--accept()/recv()/send()
48.     ZV_CALLBACK cb;
49. }zv_connect;
51. // 连接块的头
52. typedef struct zv_connblock_s{
53.     struct zv_connect_s *block;  // 指向的当前块，注意大小为 connblock_size
54.     struct zv_connblock_s *next;  // 指向的下一个连接块的头
55. }zv_connblock;
57. // 反应堆结构体
58. typedef struct zv_reactor_s{
59.     int epfd;   // epoll文件描述符
60.     // struct epoll_event events[epoll_events_size];  // 就绪事件集合
61.     struct zv_connblock_s *blockheader;  // 连接块的第一个头
62.     int blkcnt;  // 现有的连接块的总数
63. }zv_reactor;
65. // reactor初始化
66. int init_reactor(zv_reactor *reactor);
67. // reator销毁
68. void destory_reactor(zv_reactor* reactor);
69. // 服务端初始化：将端口设置为listen状态
70. int init_sever(int port);
71. // 将本地的listenfd添加进epoll
72. int set_listener(zv_reactor *reactor, int listenfd, ZV_CALLBACK cb);
73. // 创建一个新的连接块（尾插法）
74. int zv_create_connblock(zv_reactor* reactor);
75. // 根据fd从连接块中找到连接所在的位置
76. // 逻辑：整除找到所在的连接块、取余找到在连接块的位置
77. zv_connect* zv_connect_idx(zv_reactor* reactor, int fd);
78. // 运行kv存储协议
79. int kv_run_while(int argc, char *argv[]);
80. /*-------------------------------------------------------*/
83. /*-----------------------函数定义-------------------------*/
84. // reactor初始化
85. int init_reactor(zv_reactor *reactor){
86.     if(reactor == NULL) return -1;
87.     // 初始化参数
88.     memset(reactor, 0, sizeof(zv_reactor));
89.     reactor->epfd = epoll_create(1);
90.     if(reactor->epfd <= 0){
91.         printf("init reactor->epfd error: %s\n", strerror(errno));
92.         return -1;
93.     }
94.     // 为链表集合分配内存
95.     reactor->blockheader = (zv_connblock*)calloc(1, sizeof(zv_connblock));
96.     if(reactor->blockheader == NULL) return -1;
97.     reactor->blockheader->next = NULL;
98.     // 为链表集合中的第一个块分配内存
99.     reactor->blockheader->block = (zv_connect*)calloc(connblock_size, sizeof(zv_connect));
100.     if(reactor->blockheader->block == NULL) return -1;
101.     reactor->blkcnt = 1;
102.     return 0;
103. }
105. // reator销毁
106. void destory_reactor(zv_reactor* reactor){
107.     if(reactor){
108.         close(reactor->epfd);  // 关闭epoll
109.         zv_connblock* curblk = reactor->blockheader;
110.         zv_connblock* nextblk = reactor->blockheader;
111.         do{
112.             curblk = nextblk;
113.             nextblk = curblk->next;
114.             if(curblk->block) free(curblk->block);
115.             if(curblk) free(curblk);
116.         }while(nextblk != NULL);
117.     }
118. }
120. // 服务端初始化：将端口设置为listen状态
121. int init_sever(int port){
122.     // 创建服务端
123.     int sockfd = socket(AF_INET, SOCK_STREAM, 0);  // io
124.     // fcntl(sockfd, F_SETFL, O_NONBLOCK);  // 非阻塞
125.     // 设置网络地址和端口
126.     struct sockaddr_in servaddr;
127.     memset(&servaddr, 0, sizeof(struct sockaddr_in));
128.     servaddr.sin_family = AF_INET;  // IPv4
129.     servaddr.sin_addr.s_addr = htonl(INADDR_ANY);  // 0.0.0.0，任何地址都可以连接本服务器
130.     servaddr.sin_port = htons(port);  // 端口
131.     // 将套接字绑定到一个具体的本地地址和端口
132.     if(-1 == bind(sockfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr))){
133.         printf("bind failed: %s", strerror(errno));
134.         return -1;
135.     }
136.     // 将端口设置为listen（并不会阻塞程序执行）
137.     listen(sockfd, 10);  // 等待连接队列的最大长度为10
138.     printf("listen port: %d, sockfd: %d\n", port, sockfd);
139.     return sockfd;
140. }
142. // 将本地的listenfd添加进epoll
143. int set_listener(zv_reactor *reactor, int listenfd, ZV_CALLBACK cb){
144.     if(!reactor || !reactor->blockheader) return -1;
145.     // 将服务端放进连接块
146.     reactor->blockheader->block[listenfd].fd = listenfd;
147.     reactor->blockheader->block[listenfd].cb = cb;  // listenfd的回调函数应该是accept()
148.     // 将服务端添加进epoll事件
149.     struct epoll_event ev;
150.     ev.data.fd = listenfd;
151.     ev.events = EPOLLIN;
152.     epoll_ctl(reactor->epfd, EPOLL_CTL_ADD, listenfd, &ev);
153.     return 0;
154. }
156. // 创建一个新的连接块（尾插法）
157. int zv_create_connblock(zv_reactor* reactor){
158.     if(!reactor) return -1;
159.     // 初始化新的连接块
160.     zv_connblock* newblk = (zv_connblock*)calloc(1, sizeof(zv_connblock));
161.     if(newblk == NULL) return -1;
162.     newblk->block = (zv_connect*)calloc(connblock_size, sizeof(zv_connect));
163.     if(newblk->block == NULL) return -1;
164.     newblk->next = NULL;
165.     // 找到最后一个连接块
166.     zv_connblock* endblk = reactor->blockheader;
167.     while(endblk->next != NULL){
168.         endblk = endblk->next;
169.     }
170.     // 添加上新的连接块
171.     endblk->next = newblk;
172.     reactor->blkcnt++;
173.     return 0;
174. }
176. // 根据fd从连接块中找到连接所在的位置
177. // 逻辑：整除找到所在的连接块、取余找到在连接块的位置
178. zv_connect* zv_connect_idx(zv_reactor* reactor, int fd){
179.     if(!reactor) return NULL;
180.     // 计算fd应该在的连接块
181.     int blkidx = fd / connblock_size;
182.     while(blkidx >= reactor->blkcnt){
183.         zv_create_connblock(reactor);
184.         // printf("create a new connblk!\n");
185.     }
186.     // 找到这个连接块
187.     zv_connblock* blk = reactor->blockheader;
188.     int i = 0;
189.     while(++i < blkidx){
190.         blk = blk->next;
191.     }
192.     return &blk->block[fd % connblock_size];
193. }
195. // 回调函数：建立连接
196. // fd：服务端监听端口listenfd
197. // event：没用到，但是回调函数的常用格式
198. // arg：应该是reactor*
199. int accept_cb(int fd, int event, void* arg){
200.     // 与客户端建立连接
201.     struct sockaddr_in clientaddr;  // 连接到本服务器的客户端信息
202.     socklen_t len_sockaddr = sizeof(clientaddr);
203.     int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len_sockaddr);
204.     if(clientfd < 0){
205.         printf("accept() error: %s\n", strerror(errno));
206.         return -1;
207.     }
208.     // 将连接添加进连接块
209.     zv_reactor* reactor = (zv_reactor*)arg;
210.     zv_connect* conn = zv_connect_idx(reactor, clientfd);
211.     conn->fd = clientfd;
212.     conn->cb = recv_cb;
213.     conn->next_len = max_buffer_len;
214.     conn->rcount = 0;
215.     conn->wcount = 0;
216.     // 将客户端添加进epoll事件
217.     struct epoll_event ev;
218.     ev.data.fd = clientfd;
219.     ev.events = EPOLLIN;  // 默认水平触发（有数据就触发）
220.     epoll_ctl(reactor->epfd, EPOLL_CTL_ADD, clientfd, &ev);
221.     printf("connect success! sockfd:%d, clientfd:%d\n", fd, clientfd);
222. }
224. // 回调函数：接收数据
225. int recv_cb(int clientfd, int event, void* arg){
226.     zv_reactor* reactor = (zv_reactor*)arg;
227.     zv_connect* conn = zv_connect_idx(reactor, clientfd);
229.     int recv_len = recv(clientfd, conn->rbuffer+conn->rcount, conn->next_len, 0);  // 由于当前fd可读所以没有阻塞
230.     if(recv_len < 0){
231.         printf("recv() error: %s\n", strerror(errno));
232.         close(clientfd);
233.         // return -1;
234.         exit(0);
235.     }else if(recv_len == 0){
236.         // 重置对应的连接块
237.         conn->fd = -1;
238.         conn->rcount = 0;
239.         conn->wcount = 0;
240.         // 从epoll监听事件中移除
241.         epoll_ctl(reactor->epfd, EPOLL_CTL_DEL, clientfd, NULL);
242.         // 关闭连接
243.         close(clientfd);
244.         printf("close clientfd:%d\n", clientfd);
245.     }else if(recv_len > 0){
246.         conn->rcount += recv_len;
247.         // conn->next_len = *(short*)conn->rbuffer;  // 从tcp协议头中获取数据长度，假设前两位是长度
248.         
249.         // 处理接收到的字符串，并将需要发回的信息存储在缓冲区中
250.         printf("recv clientfd:%d, len:%d, mess: %s\n", clientfd, recv_len, conn->rbuffer);
251.         // conn->rcount = kv_protocal(conn->rbuffer, max_buffer_len);
253.         // 将kv存储的回复消息（rbuffer）拷贝给wbuffer
254.         // printf("msg:%s len:%ld\n", msg, strlen(msg));
255.         memset(conn->wbuffer, '\0', max_buffer_len);
256.         memcpy(conn->wbuffer, conn->rbuffer, conn->rcount);
257.         conn->wcount = conn->rcount;
258.         memset(conn->rbuffer, 0, max_buffer_len);
259.         conn->rcount = 0;
261.         // 将本连接更改为epoll写事件
262.         conn->cb = send_cb;
263.         struct epoll_event ev;
264.         ev.data.fd = clientfd;
265.         ev.events = EPOLLOUT;
266.         epoll_ctl(reactor->epfd, EPOLL_CTL_MOD, clientfd, &ev);
267.     }
268.     return 0;
269. }
271. // 回调函数：发送数据
272. int send_cb(int clientfd, int event, void* arg){
273.     zv_reactor* reactor = (zv_reactor*)arg;
274.     zv_connect* conn = zv_connect_idx(reactor, clientfd);
276.     int send_len = send(clientfd, conn->wbuffer, conn->wcount, 0);
277.      if(send_len < 0){
278.         printf("send() error: %s\n", strerror(errno));
279.         close(clientfd);
280.         return -1;
281.      }
282.     memset(conn->wbuffer, 0, conn->next_len);
283.     conn->wcount -= send_len;
285.     // 发送完成后将本连接再次更改为读事件
286.     conn->cb = recv_cb;
287.     struct epoll_event ev;
288.     ev.data.fd = clientfd;
289.     ev.events = EPOLLIN;
290.     epoll_ctl(reactor->epfd, EPOLL_CTL_MOD, clientfd, &ev);
291.     return 0;
292. }
295. // 运行kv存储协议
296. int kv_run_while(int argc, char *argv[]){
297.     // 创建管理连接的反应堆
298.     // zv_reactor reactor;
299.     zv_reactor *reactor = (zv_reactor*)malloc(sizeof(zv_reactor));
300.     init_reactor(reactor);
301.     // 服务端初始化
302.     int start_port = atoi(argv[1]);
303.     for(int i=0; i<listen_port_count; i++){
304.         int sockfd = init_sever(start_port+i);
305.         set_listener(reactor, sockfd, accept_cb);  // 将sockfd添加进epoll
306.     }
307.     printf("init finish, listening connet...\n");
308.     // 开始监听事件
309.     struct epoll_event events[epoll_events_size] = {0};  // 就绪事件集合
310.     while(1){
311.         // 等待事件发生
312.         int nready = epoll_wait(reactor->epfd, events, epoll_events_size, -1);  // -1等待/0不等待/正整数则为等待时长
313.         if(nready == -1){
314.             printf("epoll_wait error: %s\n", strerror(errno));
315.             break;
316.         }else if(nready == 0){
317.             continue;
318.         }else if(nready > 0){
319.             // printf("process %d epoll events...\n", nready);
320.             // 处理所有的就绪事件
321.             int i = 0;
322.             for(i=0; i<nready; i++){
323.                 int connfd = events[i].data.fd;
324.                 zv_connect* conn = zv_connect_idx(reactor, connfd);
326.                 // 回调函数和下面的的逻辑实现了数据回环
327.                 if(EPOLLIN & events[i].events){
328.                     conn->cb(connfd, events[i].events, reactor);
329.                 }
330.                 if(EPOLLOUT & events[i].events){
331.                     conn->cb(connfd, events[i].events, reactor);
332.                 }
333.             }
334.         }
335.     }
336.     destory_reactor(reactor);
337.     return 0;
338. }
340. int main(int argc, char *argv[]){
341.     if(argc != 2){
342.         printf("please enter port! e.x. 9999.\n");
343.         return -1;
344.     }
345.     // 初始化存储引擎
346.     // kv_engine_init();
348.     // 运行kv存储
349.     kv_run_while(argc, argv);
350.    
351.     // 销毁存储引擎
352.     // kv_engine_desy();
353.     return 0;
354. }
355. /*-------------------------------------------------------*/

复制代码

  注：只有251行、346行、352行是后续和kv存储有关的接口函数。可以发现“网络层”和“协议层”被很好的隔脱离来。
   

    图3 验证网络数据回环  可以看到上述实现了网络数据回环的功能。并且进一步来说，假如客户端利用浏览器（http协议）对该端口进行访问，那么对接收到的数据包进行http协议拆包，根据其请求的内容返回相应的信息（如html文件），那么就是我们所熟知的“web服务器”了。为什么是“烂大街”啊，一代比一代卷是吧