KVstore :键值映射存储服务器

灌篮少年  金牌会员 | 2024-6-13 20:08:01 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 946|帖子 946|积分 2838

概述：本文介绍kv存储服务，所谓kv即key-value映射，用户存储键值对，提供：1.根据键查找值 2.根据键修改值 3.根据键删除值

效果：kv存储是运行在服务器上的一个历程，客户端通过套接字与服务器上的kvstore历程进行通信，客户端发送由协议规定的请求例如 SET name01 wjq ，kvstore服务器接收到请求并解析，复兴效果 SUCCESS； 又例如客户端发送 GET name01 ，接收到服务端的复兴 wjq

实现思路：

1.首先我们必要做到kvstore与客户端通信，这里使用tcp，也就是说设计之初kvstore就是一个支持百万级并发毗连的tcp服务器：这里使用一个reactor模型，直接附上代码，tcp服务器不在本文讲解范围内
  4. #include <sys/socket.h>
  5. #include <errno.h>
  6. #include <netinet/in.h>
  8. #include <stdio.h>
  9. #include <string.h>
  10. #include <unistd.h>
  12. #include <pthread.h>
  13. #include <sys/poll.h>
  14. #include <sys/epoll.h>
  15. #include <sys/time.h>
  18. #include "kvstore.h"
  22. // listenfd
  23. // EPOLLIN -->
  24. int accept_cb(int fd);
  25. // clientfd
  26. //
  27. int recv_cb(int fd);
  28. int send_cb(int fd);
  30. // conn, fd, buffer, callback
  33. int epfd = 0;
  34. struct conn_item connlist[1048576] = {0}; // 1024  2G     2 * 512 * 1024 * 1024
  35. // list
  36. struct timeval zvoice_king;
  37. //
  38. // 1000000
  40. #define TIME_SUB_MS(tv1, tv2)  ((tv1.tv_sec - tv2.tv_sec) * 1000 + (tv1.tv_usec - tv2.tv_usec) / 1000)
  43. int set_event(int fd, int event, int flag) {
  45.         if (flag) { // 1 add, 0 mod
  46.                 struct epoll_event ev;
  47.                 ev.events = event ;
  48.                 ev.data.fd = fd;
  49.                 epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);
  50.         } else {
  51.        
  52.                 struct epoll_event ev;
  53.                 ev.events = event;
  54.                 ev.data.fd = fd;
  55.                 epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
  56.         }
  58.        
  60. }
  62. int accept_cb(int fd) {
  64.         struct sockaddr_in clientaddr;
  65.         socklen_t len = sizeof(clientaddr);
  66.        
  67.         int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len);
  68.         if (clientfd < 0) {
  69.                 return -1;
  70.         }
  71.         set_event(clientfd, EPOLLIN, 1);
  73.         connlist[clientfd].fd = clientfd;
  74.         memset(connlist[clientfd].rbuffer, 0, BUFFER_LENGTH);
  75.         connlist[clientfd].rlen = 0;
  76.         memset(connlist[clientfd].wbuffer, 0, BUFFER_LENGTH);
  77.         connlist[clientfd].wlen = 0;
  78.        
  79.         connlist[clientfd].recv_t.recv_callback = recv_cb;
  80.         connlist[clientfd].send_callback = send_cb;
  82.         if ((clientfd % 1000) == 999) {
  83.                 struct timeval tv_cur;
  84.                 gettimeofday(&tv_cur, NULL);
  85.                 int time_used = TIME_SUB_MS(tv_cur, zvoice_king);
  87.                 memcpy(&zvoice_king, &tv_cur, sizeof(struct timeval));
  88.                
  89.                 printf("clientfd : %d, time_used: %d\n", clientfd, time_used);
  90.         }
  92.         return clientfd;
  93. }
  95. int recv_cb(int fd) { // fd --> EPOLLIN
  97.         char *buffer = connlist[fd].rbuffer;
  98.         int idx = connlist[fd].rlen;
  99.        
  100.         int count = recv(fd, buffer, BUFFER_LENGTH, 0);
  101.         if (count == 0) {
  102.                 printf("disconnect\n");
  104.                 epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);               
  105.                 close(fd);
  106.                
  107.                 return -1;
  108.         }
  110.        
  111.         connlist[fd].rlen = count;
  114.         kvstore_request(&connlist[fd]);
  115.         connlist[fd].wlen = strlen(connlist[fd].wbuffer);
  117.         set_event(fd, EPOLLOUT, 0);   
  119.        
  120.         return count;
  121. }
  124. int send_cb(int fd) {
  126.         char *buffer = connlist[fd].wbuffer;
  127.         int idx = connlist[fd].wlen;
  129.         int count = send(fd, buffer, idx, 0);
  131.         set_event(fd, EPOLLIN, 0);
  133.         return count;
  134. }
  139. int init_server(unsigned short port) {
  141.         int sockfd = socket(AF_INET, SOCK_STREAM, 0);
  143.         struct sockaddr_in serveraddr;
  144.         memset(&serveraddr, 0, sizeof(struct sockaddr_in));
  146.         serveraddr.sin_family = AF_INET;
  147.         serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
  148.         serveraddr.sin_port = htons(port);
  150.         if (-1 == bind(sockfd, (struct sockaddr*)&serveraddr, sizeof(struct sockaddr))) {
  151.                 perror("bind");
  152.                 return -1;
  153.         }
  155.         listen(sockfd, 10);
  157.         return sockfd;
  158. }
  163. int epoll_entry(void) {
  165.         int port_count = 20;
  166.         unsigned short port = 2048;
  167.         int i = 0;
  169.        
  170.         epfd = epoll_create(1); // int size
  172.         for (i = 0;i < port_count;i ++) {
  173.                 int sockfd = init_server(port + i);  // 2048, 2049, 2050, 2051 ... 2057
  174.                 connlist[sockfd].fd = sockfd;
  175.                 connlist[sockfd].recv_t.accept_callback = accept_cb;
  176.                 set_event(sockfd, EPOLLIN, 1);
  177.         }
  179.         gettimeofday(&zvoice_king, NULL);
  181.         struct epoll_event events[1024] = {0};
  182.        
  183.         while (1) { // mainloop();
  185.                 int nready = epoll_wait(epfd, events, 1024, -1); //
  187.                 int i = 0;
  188.                 for (i = 0;i < nready;i ++) {
  190.                         int connfd = events[i].data.fd;
  191.                         if (events[i].events & EPOLLIN) { //
  193.                                 int count = connlist[connfd].recv_t.recv_callback(connfd);
  194.                                 //printf("recv count: %d <-- buffer: %s\n", count, connlist[connfd].rbuffer);
  196.                         } else if (events[i].events & EPOLLOUT) {
  197.                                 // printf("send --> buffer: %s\n",  connlist[connfd].wbuffer);
  198.                                
  199.                                 int count = connlist[connfd].send_callback(connfd);
  200.                         }
  202.                 }
  204.         }
  207.         //getchar();
  208.         //close(clientfd);
  210. }
复制代码
函数epoll_entry实现了与客户端之间的通信，并通过kvstore_request(&connlist[fd])这个函数实现了处置惩罚客户端请求，并将处置惩罚效果发送给客户端

2.kvstore存储引擎的实现

概要：由于服务器要将客户端请求存储的内容存储起来，有两种方式，一是存储到数据库，二是存储到服务端本地
为了简朴实现业务，本文使用存储到本地进行讲解，接纳的数据结构是哈希表
先介绍哈希表的实现以及为kvstore封装的接口：
  1. /*
  2. * 单线程版本，没有做线程安全！
  3. *
  4. */
  7. #include <stdio.h>
  8. #include <string.h>
  9. #include <stdlib.h>
  10. #include <pthread.h>
  12. #include "kvstore.h"
  15. #define MAX_KEY_LEN        128
  16. #define MAX_VALUE_LEN        512
  19. #define MAX_TABLE_SIZE        102400
  21. #define ENABLE_POINTER_KEY        1
  24. typedef struct hashnode_s { // hash node
  26. #if ENABLE_POINTER_KEY
  27.     char *key;
  28.     char *value;
  29. #else
  30.     char key[MAX_KEY_LEN];
  31.     char value[MAX_VALUE_LEN];
  32. #endif
  33.     struct hashnode_s *next;
  35. } hashnode_t;
  38. typedef struct hashtable_s { // hash table
  40.     hashnode_t **nodes; // hashnode_t * 类型的 *nodes，也就是存放着hashnode_t类型的指针的数组nodes
  42.     int max_slots;
  43.     int count;
  45. } hashtable_t;
  48. hashtable_t Hash;
  52. static int _hash(char *key, int size) { // hash函数，使用key确定hash值
  54.     if (!key) return -1;
  56.     int sum = 0;
  57.     int i = 0;
  59.     while (key[i] != 0) { // 使用ASCII计算hash值，由于key是字符数组，该方法通用
  60.         sum += key[i];
  61.         i ++;
  62.     }
  64.     return sum % size; // 返回hash值
  65. }
  68. hashnode_t *_create_node(char *key, char *value) {
  70.     hashnode_t *node = (hashnode_t *)kvstore_malloc(sizeof(hashnode_t));
  71.     if (!node) return NULL; // malloc filed
  73. #if ENABLE_POINTER_KEY
  75.     // 为节点的成员分配空间
  76.     node->key = kvstore_malloc(strlen(key) + 1);
  77.     if (!node->key) {
  78.         kvstore_free(node); // node分配成功但key失败
  79.         return NULL;
  80.     }
  81.     strcpy(node->key, key);
  83.     node->value = kvstore_malloc(strlen(value) + 1);
  84.     if (!node->value) {
  85.         kvstore_free(node->key); // node和key分配成功但value失败
  86.         kvstore_free(node);
  87.         return NULL;
  88.     }
  89.     strcpy(node->value, value);
  91. #else
  93.         strncpy(node->key, key, MAX_KEY_LEN);
  94.         strncpy(node->value, value, MAX_VALUE_LEN);
  95.        
  96. #endif
  97.     // 初始化 next
  98.     node->next = NULL;
  100.     return node;
  101. }
  105. int init_hashtable(hashtable_t *hash) {
  107.     if (!hash) return -1;
  109.     hash->nodes = (hashnode_t**)kvstore_malloc(sizeof(hashnode_t *) * MAX_TABLE_SIZE);
  110.     if (!hash->nodes) return -1;
  112.     hash->max_slots = MAX_TABLE_SIZE;
  113.     hash->count = 0;
  115.     return 0;
  116. }
  118. void dest_hashtable(hashtable_t *hash) { // 销毁哈希表
  120.     if (!hash) return;
  122.     // 遍历释放数组中所有链表
  123.     int i = 0;
  124.     for (i = 0; i < hash->max_slots; i++) {
  125.         hashnode_t *node = hash->nodes[i];
  127.         while (node != NULL) {
  128.             hashnode_t *tmp = node; // 保存当前节点
  129.             node = node->next; // 移动到下一个节点
  130.             hash->nodes[i] = node; // 更新头指针，在这段代码中没有作用
  132.             kvstore_free(tmp); // 释放当前节点
  134.         }
  135.     }
  137.     kvstore_free(hash->nodes); // 释放哈希表的数组成员
  138. }
  141. int put_kv_hashtable(hashtable_t *hash, char *key, char *value) {
  143.     if (!hash || !key || !value) return -1;
  145.     int idx = _hash(key, MAX_TABLE_SIZE); // 哈希值作为数组下标
  147.     hashnode_t *node = hash->nodes[idx]; // 获取正确数组位置的头指针
  148. #if 1
  149.     while (node != NULL) {    // 如果已经存在，直接退出，不重复插入
  150.         if (strcmp(node->key, key) == 0) {
  151.             return 1;
  152.         }
  153.         node = node->next;
  154.     }
  155. #endif
  157.     hashnode_t *new_node = _create_node(key, value);
  158.     // 头插法
  159.     new_node->next = hash->nodes[idx];
  160.     hash->nodes[idx] = new_node; // 更新头节点指针
  162.     hash->count ++;
  164.     return 0;
  165. }
  168. char *get_kv_hashtable(hashtable_t *hash, char *key) { // search
  170.     if (!hash || !key) return NULL;
  172.     int idx = _hash(key, MAX_TABLE_SIZE);
  174.     hashnode_t *node = hash->nodes[idx]; // 确定数组索引
  176.     while (node != NULL) { // 遍历查找
  178.         if (strcmp(node->key, key) == 0) {
  179.             return node->value;
  180.         }
  182.         node = node->next;
  183.     }
  185.     return NULL;
  186. }
  189. int count_kv_hashtable(hashtable_t *hash) {
  190.         return hash->count;
  191. }
  194. int delete_kv_hashtable(hashtable_t *hash, char *key) { // 根据key删除节点
  196.     if (!hash || !key) return -1;
  198.     int idx = _hash(key, MAX_TABLE_SIZE); // 哈希值作为索引
  199.     // 先判断头指针
  200.     hashnode_t *head = hash->nodes[idx];
  201.     if (head == NULL) return -1;
  203.     // 遍历链表
  204.     hashnode_t *cur = hash->nodes;
  205.     hashnode_t *prev = NULL;
  206.     while (cur != NULL) {
  208.         if (strcmp(cur->key, key) == 0) break;
  209.         prev = cur;
  210.         cur = cur->next;
  211.     }
  213.     if (cur == NULL) return -1; // 没找到
  215.     if (prev == NULL) { // 如果要删除的是头节点
  216.         hash->nodes[idx] = cur->next; // 删除cur
  217.     } else { // 不是头节点
  218.         prev->next = cur->next; // 删除cur
  219.     }
  221. // 释放cur节点的空间
  222. #if ENABLE_POINTER_KEY
  223.         if (cur->key) {
  224.                 kvstore_free(cur->key);
  225.         }
  226.         if (cur->value) {
  227.                 kvstore_free(cur->value);
  228.         }
  229.         kvstore_free(cur);
  230. #else
  231.         free(cur);
  232. #endif
  233.         hash->count --; // 更新count
  235.     return 0;
  236. }
  239. int exit_kv_hashtable(hashtable_t *hash, char *key) { // 判断是否存在该key的映射value
  241.     char *value = get_kv_hashtable(hash, key);
  242.     if (value) return 1;
  243.     else return 0;
  244. }
  248. int kvs_hash_modify(hashtable_t *hash, char *key, char *value) { // 先查找key再修改value
  250.     if (!hash || !key || !value) return -1;
  252.     int idx = _hash(key, MAX_TABLE_SIZE);
  254.     hashnode_t *node = hash->nodes[idx];
  256.     while (node != NULL) {
  258.         if (strcmp(node->key, key) == 0) {
  259.             // 先释放原空间，避免内存泄漏
  260.             kvstore_free(node->value); // 释放原value指向的空间
  261.             node->value = NULL; // 避免使用悬空指针
  262.             // 新分配空间
  263.             node->value = kvstore_malloc(strlen(value) + 1);
  264.             if (node->value) { // 分配成功
  265.                 strcpy(node->value, value);
  266.                 return 0;
  267.             } else
  268.                 assert(0);
  269.         }
  271.         node = node->next;
  272.     }
  274.     return -1;
  275. }
  278. int kvs_hash_count(hashtable_t *hash) {
  279.         return hash->count;
  280. }
  284. // 再封装一层接口：使用第三方库时，对库函数进行一层封装，适配自己的代码，
  285. // 排查问题或更新迭代时只需要修改这一层接口的内容就行，不需要在源代码主体上修改，相当于做了一层隔离
  287. int kvstore_hash_craete(hashtable_t *hash) {
  289.     return init_hashtable(hash);
  290. }
  293. void kvstore_hash_destory(hashtable_t *hash) {
  295.         return dest_hashtable(hash);
  297. }
  300. int kvs_hash_set(hashtable_t *hash, char *key, char *value) {
  302.         return put_kv_hashtable(hash, key, value);
  304. }
  307. char *kvs_hash_get(hashtable_t *hash, char *key) {
  309.         return get_kv_hashtable(hash, key);
  311. }
  314. int kvs_hash_delete(hashtable_t *hash, char *key) {
  316.         return delete_kv_hashtable(hash, key);
  318. }
复制代码
对于哈希表的设计与实现，注释说的很清晰了，末了封装的接口是用在接下来的kvstore主程序中的：
3.kvstore主体

概要：这份代码集成了前面的tcp服务epoll_entry、存储组件哈希表以及末了要介绍的：对客户端请求进行解析处置惩罚的组件
先介绍kvstore主程序：
  1. int init_kvengine(void) {
  3.         kvstore_hash_create(&Hash);
  4. }
  7. int exit_kvengine(void) {
  8.    
  9.         kvstore_hash_destory(&Hash);
  11. }
  17. int main() {
  20.         init_kvengine(); // 创建存储引擎，这里是哈希表
  21.        
  22.         epoll_entry();  // 启动tcp服务器，处理并回复客户端请求
  24.         exit_kvengine(); // 销毁哈希表
  26. }
复制代码
而这里调用的init_kvengine();现实上就是前面的哈希表代码中的：
  1. int init_hashtable(hashtable_t *hash) {
  3.     if (!hash) return -1;
  5.     hash->nodes = (hashnode_t**)kvstore_malloc(sizeof(hashnode_t *) * MAX_TABLE_SIZE);
  6.     if (!hash->nodes) return -1;
  8.     hash->max_slots = MAX_TABLE_SIZE;
  9.     hash->count = 0;
  11.     return 0;
  12. }
复制代码
4.请求解析

我们对于kvstore主程序中的存储引擎、tcp服务都介绍完了，接下来介绍最核心的请求解析函数：
这两个函数位于epoll_entry的kvstore_request(&connlist[fd])函数中：
  1. int kvstore_request(struct conn_item *item) {
  3.     char *msg = item->rbuffer;
  4.     char *tokens[KVSTORE_MAX_TOKENS];
  6.     int count = kvstore_split_token(msg, tokens); // 解析请求
  8.     kvstore_parser_protocol(item, tokens, count); // 生成回复内容
  10.     return 0;
  11. }
复制代码
这个函数做到了对用户请求的解析以及复兴，而依靠的是以下两个函数：
解析请求：

  1. int kvstore_split_token(char *msg, char **tokens) { // 将msg字符串进行分割，结果保存在tokens字符串数字里
  3.     if (msg == NULL || tokens == NULL) return -1; // 参数检查
  5.     int idx = 0;
  7.     char *token = strtok(msg, " "); // 对msg按空格“ ”进行分割，返回第一个子字符串
  9.     while (token != NULL) { // 获取剩余的子字符串
  10.         tokens[idx++] = token; // 将子字符串保存在字符串数组里
  11.         token = strtok(NULL, " "); // 固定写法，依次获取除第一个外，剩余的子字符串
  12.     }
  14.     return idx; // 返回子字符串的个数
  15. }
复制代码
我们能对用户请求按空格进行分割的原因是，kvstore规定了应用层协议，只有按协议规定发送的请求才气被正确处置惩罚，就像linux shell 中的下令的名称以及使用方法一样
处置惩罚并复兴：

  1. int kvstore_parser_protocol(struct conn_item *item, char **tokens, int count) {
  3.     if (item == NULL || tokens[0] == NULL || count == 0) return -1; // 检查参数
  5.     char *msg = item->wbuffer; // 获取写缓冲区
  6.         memset(msg, 0, BUFFER_LENGTH);
  8.     // 对用户的命令的解析结果， 例如 SET name wjq 解析结果如下：
  9.     char *command = tokens[0];  // SET
  10.         char *key = tokens[1];      // name
  11.         char *value = tokens[2];    // wjq
  12.    
  14.     int cmd = KVS_CMD_START;
  16.    
  17.     for (cmd = KVS_CMD_START; cmd < KVS_CMD_SIZE; cmd++) { // 查找比对tokens里的命令
  19.         if (strcmp(commands[cmd], command) == 0) {
  20.             break; // 找到了或者不存在
  21.         }
  22.     }
  24.     // 匹配命令并回复结果
  25.     switch (cmd) {
  26.       
  27.         case KVS_CMD_HSET: {    // SET :添加
  29.                         int res = kvstore_hash_set(key, value); // 调用哈希表的函数
  30.                         if (!res) {
  31.                                 snprintf(msg, BUFFER_LENGTH, "SUCCESS");
  32.                         } else {
  33.                                 snprintf(msg, BUFFER_LENGTH, "FAILED");
  34.                         }
  35.                         break;
  36.                 }
  37.                
  38.                 case KVS_CMD_HGET: {   // GET ：查询
  40.                         char *val = kvstore_hash_get(key); // 调用哈希表提供的接口
  41.                         if (val) {
  42.                                 snprintf(msg, BUFFER_LENGTH, "%s", val);
  43.                         } else {
  44.                                 snprintf(msg, BUFFER_LENGTH, "NO EXIST");
  45.                         }
  46.                        
  47.                         break;
  48.                 }
  49.                 case KVS_CMD_HDEL: { // DEL ： 删除
  51.                         int res = kvstore_hash_delete(key);
  52.                         if (res < 0) {  // server
  53.                                 snprintf(msg, BUFFER_LENGTH, "%s", "ERROR");
  54.                         } else if (res == 0) {
  55.                                 snprintf(msg, BUFFER_LENGTH, "%s", "SUCCESS");
  56.                         } else {
  57.                                 snprintf(msg, BUFFER_LENGTH, "NO EXIST");
  58.                         }
  59.                        
  60.                         break;
  61.                 }
  62.         case KVS_CMD_HMOD: { // MOD : 修改
  64.                         int res = kvstore_hash_modify(key, value);
  65.                         if (res < 0) {  // server
  66.                                 snprintf(msg, BUFFER_LENGTH, "%s", "ERROR");
  67.                         } else if (res == 0) {
  68.                                 snprintf(msg, BUFFER_LENGTH, "%s", "SUCCESS");
  69.                         } else {
  70.                                 snprintf(msg, BUFFER_LENGTH, "NO EXIST");
  71.                         }
  72.                        
  73.                         break;
  74.                 }
  76.                 case KVS_CMD_HCOUNT: { // COUNT ： 查询数量
  77.                         int count = kvstore_hash_count();
  78.                         if (count < 0) {  // server
  79.                                 snprintf(msg, BUFFER_LENGTH, "%s", "ERROR");
  80.                         } else {
  81.                                 snprintf(msg, BUFFER_LENGTH, "%d", count);
  82.                         }
  83.                         break;
  84.                 }
  85.                
  86.                 default: {
  87.                         printf("cmd: %s\n", commands[cmd]);
  88.                         assert(0);
  89.                 }
  90.                
  91.     }
  93. }
复制代码
可以看到解析查询的过程就是将用户按我们指定协议输入的请求，分成几段，为每一条请求进行一次解析、处置惩罚
增删改查用到了哈希表这个数据结构提供的函数，而只有按空格将字符串分割这个函数是我们自行设计的，难度并不大
至此，kvstore的设计实现已经全部完成
保举学习https://xxetb.xetslk.com/s/p5Ibb

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

灌篮少年

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表