KV存储引擎 众所周知,非关系型数据库redis,以及levedb,rockdb其核心存储引擎的数据结构就是跳表。
本项目就是基于跳表实现的轻量级键值型存储引擎,使用C++实现。插入数据、删除数据、查询数据、数据展示、数据落盘、文件加载数据,以及数据库大小显示。
在随机写读情况下,该项目每秒可处理啊请求数(QPS): 24.39w,每秒可处理读请求数(QPS): 18.41w
项目存储文件
提供接口
存储引擎数据表现
跳表树高:18
采用随机插入数据测试:
插入数据规模(万条) 耗时(秒) 10 0.316763 50 1.86778 100 4.10648
每秒可处理写请求数(QPS): 24.39w
取数据规模(万条) 耗时(秒) 10 0.47148 50 2.56373 100 5.43204
每秒可处理读请求数(QPS): 18.41w
make // complie demo main.cpp ./bin/main // run
如果想自己写程序使用这个kv存储引擎,只需要在你的CPP文件中include skiplist.h 就可以了。
可以运行如下脚本测试kv存储引擎的性能(当然你可以根据自己的需求进行修改)
sh stress_test_start.sh
待优化
项目涉及知识
代码阅读 从main函数开始阅读,然后跳到头文件,最后压力测试 SkipList<int, std::string> skipList(6); 定义了一个6层高度的跳表
头文件全局变量
node节点类 跳表类
该项目的跳表对Redis的跳表结构进行了一定的简化 Redis跳表 跳表数据结构: 代码实现: 本项目的跳表
template<typename K, typename V> int SkipList<K, V>::insert_element(const K key, const V value) { mtx.lock(); Node<K, V> *current = this->_header; // update 是个节点数组,用于后续插入位置的前向链接 和 后向链接 Node<K, V> *update[_max_level+1]; memset(update, 0, sizeof(Node<K, V>*)*(_max_level+1)); // 寻找插入位置,并保存插入位置前面的节点 // for进行高度遍历,while 行方向遍历 for(int i = _skip_list_level; i >= 0; i--) { while(current->forward[i] != NULL && current->forward[i]->get_key() < key) { current = current->forward[i]; } update[i] = current; } // 定位到 == key 或者 > key 的 节点 current = current->forward[0]; // 如果current节点存在且和key相等,提示,并解锁 if (current != NULL && current->get_key() == key) { std::cout << "key: " << key << ", exists" << std::endl; mtx.unlock(); return 1; } //否则,需要在update[0]和current node节点之间插入 [节点] if (current == NULL || current->get_key() != key ) { // 随机生成节点的高度 int random_level = get_random_level(); // 如果随机高度比当前的跳表的高度大,update数组在多余高出来的部分保存头节点指针 if (random_level > _skip_list_level) { for (int i = _skip_list_level+1; i < random_level+1; i++) { update[i] = _header; } _skip_list_level = random_level; } // 用随机生成的高度 创建 insert node Node<K, V>* inserted_node = create_node(key, value, random_level); // node节点 forword 指向 所在位置后面的节点指针 // node节点 所在位置前面的节点 指向node for (int i = 0; i <= random_level; i++) { inserted_node->forward[i] = update[i]->forward[i]; update[i]->forward[i] = inserted_node; } std::cout << "Successfully inserted key:" << key << ", value:" << value << std::endl; _element_count ++; } mtx.unlock(); return 0; }
template<typename K, typename V> void SkipList<K, V>::delete_element(K key) { mtx.lock(); Node<K, V> *current = this->_header; // update数组 用于保存 删除节点前面的节点 ,用于删除节点后的指向链接,步骤和插入节点类似 Node<K, V> *update[_max_level+1]; memset(update, 0, sizeof(Node<K, V>*)*(_max_level+1)); // 寻找要删除的节点,并将该节点前面的节点保存到update数组中 for (int i = _skip_list_level; i >= 0; i--) { while (current->forward[i] !=NULL && current->forward[i]->get_key() < key) { current = current->forward[i]; } update[i] = current; } // ==key 或者 > key 的元素节点 current = current->forward[0]; // 如果相等进行下面的操作,不相等直接解锁 return if (current != NULL && current->get_key() == key) { // 从低级别开始删除每层的节点, for (int i = 0; i <= _skip_list_level; i++) { // 如果在第i曾,其下一个节点不是所要删除的节点直接break if (update[i]->forward[i] != current) break; // 如果是要删除的节点 重新设置指针的指向,将其指向下一个节点的位置 update[i]->forward[i] = current->forward[i]; } // 减少没有元素的层,更新跳表的level while (_skip_list_level > 0 && _header->forward[_skip_list_level] == 0) { _skip_list_level --; } std::cout << "Successfully deleted key "<< key << std::endl; _element_count --; } mtx.unlock(); return; }
// Search for element in skip list /* +------------+ | select 60 | +------------+ level 4 +-->1+ 100 | | level 3 1+-------->10+------------------>50+ 70 100 | | level 2 1 10 30 50| 70 100 | | level 1 1 4 10 30 50| 70 100 | | level 0 1 4 9 10 30 40 50+-->60 70 100 */ template<typename K, typename V> bool SkipList<K, V>::search_element(K key) { std::cout << "search_element-----------------" << std::endl; Node<K, V> *current = _header; // for循环高度遍历,while循环水平遍历每一层。 for (int i = _skip_list_level; i >= 0; i--) { while (current->forward[i] && current->forward[i]->get_key() < key) { current = current->forward[i]; } } // 找到 > 或者 = key的 节点 current = current->forward[0]; // 如果存在且相等,成功找到 if (current and current->get_key() == key) { std::cout << "Found key: " << key << ", value: " << current->get_value() << std::endl; return true; } // 否则 没有找到 进行提示 std::cout << "Not Found Key:" << key << std::endl; return false; }
github网址:
pthread_t threads[NUM_THREADS]; pthread_create(&threads[i], NULL, insertElement, (void *)i); void *insertElement(void* threadid) { long tid; tid = (long)threadid; std::cout << tid << std::endl; int tmp = TEST_COUNT/NUM_THREADS; for (int i=tid*tmp, count=0; count<tmp; i++) { count++; skipList.insert_element(rand() % TEST_COUNT, "a"); } pthread_exit(NULL); }
虚拟机测试: 项目官方测试:
知识补充: 4. 取消内存对齐 5.
异步日志系统主要涉及了两个模块,一个是日志模块,一个是阻塞队列模块,其中加入阻塞队列模块主要是解决异步写入日志做准备.
自定义阻塞队列 单例模式创建日志 异步日志
/************************************************************* *循环数组实现的阻塞队列,m_back = (m_back + 1) % m_max_size; *线程安全,每个操作前都要先加互斥锁,操作完后,再解锁 **************************************************************/
private: locker m_mutex; // 循环数组加锁 cond m_cond; // 条件变量通知 T *m_array; // 循环数组 int m_size; // 已经有的size int m_max_size; // 最大的size int m_front; // 头部元素 int m_back; // 尾部元素
block_queue(int max_size = 1000) { if (max_size <= 0) { exit(-1); } m_max_size = max_size; m_array = new T[max_size]; m_size = 0; m_front = -1; m_back = -1; }
~block_queue() { m_mutex.lock(); if (m_array != NULL) delete [] m_array; m_mutex.unlock(); }
bool push(const T &item) { m_mutex.lock(); if (m_size >= m_max_size) { m_cond.broadcast(); m_mutex.unlock(); return false; } m_back = (m_back + 1) % m_max_size; m_array[m_back] = item; m_size++; m_cond.broadcast(); m_mutex.unlock(); return true; }
//pop时,如果当前队列没有元素,将会等待条件变量 bool pop(T &item) { m_mutex.lock(); //多个消费者的时候,这里要是用while而不是if while (m_size <= 0) { //当重新抢到互斥锁 if (!m_cond.wait(m_mutex.get())) { m_mutex.unlock(); return false; } } m_front = (m_front + 1) % m_max_size; item = m_array[m_front]; m_size--; m_mutex.unlock(); return true; }
void clear() { m_mutex.lock(); m_size = 0; m_front = -1; m_back = -1; m_mutex.unlock(); }
bool full() bool empty() bool front(T &value) bool back(T &value)
单例模式
private: char dir_name[128]; //路径名 char log_name[128]; //log文件名 int m_log_buf_size; //日志缓冲区大小 FILE *m_fp; //打开log的文件指针 char *m_buf; block_queue<string> *m_log_queue; //阻塞队列 locker m_mutex;
private: Log(); virtual ~Log();
//C++11以后,使用局部变量懒汉不用加锁 static Log *get_instance() { static Log instance; return &instance; } static void *flush_log_thread(void *args) { string single_log; //从阻塞队列中取出一个日志string,写入文件 while (m_log_queue->pop(single_log)) { m_mutex.lock(); fputs(single_log.c_str(), m_fp); m_mutex.unlock(); } } //可选择的参数有日志文件、日志缓冲区大小、最长日志条队列 bool init(const char *file_name, int log_buf_size = 8192, int max_queue_size = 1000); void write_log(int level, const char *format, ...); void flush(void);
m_log_queue = new block_queue<string>(max_queue_size); pthread_t tid; //flush_log_thread为回调函数,这里表示创建线程异步写日志 pthread_create(&tid, NULL, flush_log_thread, NULL); m_fp = fopen(log_full_name, "a");
if (!m_log_queue->full()) { m_log_queue->push(log_str); } else { m_mutex.lock(); fputs(log_str.c_str(), m_fp); m_mutex.unlock(); }
void Log::flush(void) { m_mutex.lock(); //强制刷新写入流缓冲区 fflush(m_fp); m_mutex.unlock(); }
文章浏览阅读1.6k次。安装配置gi、安装数据库软件、dbca建库见下:http://blog.csdn.net/kadwf123/article/details/784299611、检查集群节点及状态:[root@rac2 ~]# olsnodes -srac1 Activerac2 Activerac3 Activerac4 Active[root@rac2 ~]_12c查看crs状态
文章浏览阅读1.3w次,点赞45次,收藏99次。我个人用的是anaconda3的一个python集成环境,自带jupyter notebook,但在我打开jupyter notebook界面后,却找不到对应的虚拟环境,原来是jupyter notebook只是通用于下载anaconda时自带的环境,其他环境要想使用必须手动下载一些库:1.首先进入到自己创建的虚拟环境(pytorch是虚拟环境的名字)activate pytorch2.在该环境下下载这个库conda install ipykernelconda install nb__jupyter没有pytorch环境
文章浏览阅读5.2k次,点赞19次,收藏28次。选择scoop纯属意外,也是无奈,因为电脑用户被锁了管理员权限,所有exe安装程序都无法安装,只可以用绿色软件,最后被我发现scoop,省去了到处下载XXX绿色版的烦恼,当然scoop里需要管理员权限的软件也跟我无缘了(譬如everything)。推荐添加dorado这个bucket镜像,里面很多中文软件,但是部分国外的软件下载地址在github,可能无法下载。以上两个是官方bucket的国内镜像,所有软件建议优先从这里下载。上面可以看到很多bucket以及软件数。如果官网登陆不了可以试一下以下方式。_scoop-cn
文章浏览阅读4.5k次,点赞2次,收藏3次。首先要有一个color-picker组件 <el-color-picker v-model="headcolor"></el-color-picker>在data里面data() { return {headcolor: ’ #278add ’ //这里可以选择一个默认的颜色} }然后在你想要改变颜色的地方用v-bind绑定就好了,例如:这里的:sty..._vue el-color-picker
文章浏览阅读640次。基于芯片日益增长的问题,所以内核开发者们引入了新的方法,就是在内核中只保留函数,而数据则不包含,由用户(应用程序员)自己把数据按照规定的格式编写,并放在约定的地方,为了不占用过多的内存,还要求数据以根精简的方式编写。boot启动时,传参给内核,告诉内核设备树文件和kernel的位置,内核启动时根据地址去找到设备树文件,再利用专用的编译器去反编译dtb文件,将dtb还原成数据结构,以供驱动的函数去调用。firmware是三星的一个固件的设备信息,因为找不到固件,所以内核启动不成功。_exynos 4412 刷机
文章浏览阅读2w次,点赞24次,收藏42次。Linux系统配置jdkLinux学习教程,Linux入门教程(超详细)_linux配置jdk
文章浏览阅读3.3k次,点赞5次,收藏19次。xlabel('\delta');ylabel('AUC');具体符号的对照表参照下图:_matlab微米怎么输入
文章浏览阅读119次。顺序读写指的是按照文件中数据的顺序进行读取或写入。对于文本文件,可以使用fgets、fputs、fscanf、fprintf等函数进行顺序读写。在C语言中,对文件的操作通常涉及文件的打开、读写以及关闭。文件的打开使用fopen函数,而关闭则使用fclose函数。在C语言中,可以使用fread和fwrite函数进行二进制读写。 Biaoge 于2024-03-09 23:51发布 阅读量:7 ️文章类型:【 C语言程序设计 】在C语言中,用于打开文件的函数是____,用于关闭文件的函数是____。
文章浏览阅读3.4k次,点赞2次,收藏13次。跟随鼠标移动的粒子以grid(SOP)为partical(SOP)的资源模板,调整后连接【Geo组合+point spirit(MAT)】,在连接【feedback组合】适当调整。影响粒子动态的节点【metaball(SOP)+force(SOP)】添加mouse in(CHOP)鼠标位置到metaball的坐标,实现鼠标影响。..._touchdesigner怎么让一个模型跟着鼠标移动
文章浏览阅读178次。项目运行环境配置:Jdk1.8 + Tomcat7.0 + Mysql + HBuilderX(Webstorm也行)+ Eclispe(IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持)。项目技术:Springboot + mybatis + Maven +mysql5.7或8.0+html+css+js等等组成,B/S模式 + Maven管理等等。环境需要1.运行环境:最好是java jdk 1.8,我们在这个平台上运行的。其他版本理论上也可以。_基于java技术的停车场管理系统实现与设计
文章浏览阅读3.5k次。前言对于MediaPlayer播放器的源码分析内容相对来说比较多,会从Java-&amp;gt;Jni-&amp;gt;C/C++慢慢分析,后面会慢慢更新。另外,博客只作为自己学习记录的一种方式,对于其他的不过多的评论。MediaPlayerDemopublic class MainActivity extends AppCompatActivity implements SurfaceHolder.Cal..._android多媒体播放源码分析 时序图
文章浏览阅读2.4k次,点赞41次,收藏13次。java 数据结构与算法 ——快速排序法_快速排序法