欢迎光临
我们一直在努力

B树概述与简单应用示例(C#)

引言:

  天不生仲尼,万古如长夜。在计算机科学中,也有一个划时代的发明,B树(多路平衡查找树)及其变体(B树,b*树,b+树);

由德国科学家(鲁道夫·拜尔 Rudolf Bayer),美国科学家(爱德华·M·麦克特 Edward Meyers McCreight)于1970年共同发明;

B树这种数据结构特别适合用于数据库与文件系统设计中,是人类精神财富的精华部分,B树不诞生,计算机在处理大数据量计算时会变得非常困难。

 

用途:

  基本上都是软件产品最底层的,最核心的功能。

如:各种操作系统(windows,Linux,Mac)的文件系统索引,各种数据库(sqlserver、oracle、mysql、MongoDB、等等),

基本上大部分与大数据量读取有关的事务,多少都与B树家族有关,因为B树的优点太明显,特别是读取磁盘数据效率非常的高效,

查找效率O(log n),甚至在B+树中查询速度恒定,无论多少存储多少数据,查询任何一个速度都一样。简直就是天才的发明。

 

诞生的原因:

  在上世纪时期,计算机内存储器都非常的小,以KB为单位,比起现在动不动以G计算,简直小的可怜。

计算机运算数据时,数据是在内存中进行操作的,比如一些加减乘除、正删改查等。

举个简单的栗子:从一个数组 int a[1,2,3,4,5,6,7,8,9]中找出3,那非常简单;大概步骤如下:

  1、在内存中初始化这个数组

  2、获取数组指针遍历这个数组,查到3就完成

  但是这个数组很大,比如包含1亿个数字怎么办?如果数组容量大大超过内存大小,那这种比较就不现实了。现在的做法都是把文件

数据存放在外存储器,比如磁盘,U盘,光盘;然后把文件分多次的拷贝数据至内存进行操作。但是读取外存储器效率对比读取内存,

差距是非常大的,一般是百万级别的差距,差6个数量级,所以这个问题不解决一切都是空谈。

  好在操作系统在设计之初,就对读取外存储器进行了一定的优化,引入了“逻辑块”概念,当做操作文件的最小单元,而B树合理地利用这个“逻辑块”

功能开发的高效存储数据结构;在介绍B树特性之前,先来了解一下磁盘的基本工作原理。

 

磁盘简单介绍:

1)磁盘结构介绍 

 

  网上引用的两张图,将就看看,基本结构是:磁盘 > 盘面 > 磁道 > 扇区

  左边是物理图,这个大家应该都是经常见到了,一般圆形的那部分有很多层,每一层叫盘片;右边的是示意图,代表左图的一个盘面。

每个盘面有跟多环形的磁道,每个磁道有若干段扇区组成,扇区是磁盘的最小组成单元,若干段扇区组成簇(也叫磁盘块、逻辑块等)

先看看我电脑的磁盘簇与扇区大小

  可以看到我的E盘每个扇区512个字节,每个簇4096字节,这个先记下来,后边有用到

扇区是磁盘组成的最小单元,簇是虚拟出来的,主要是为了操作系统方便读写磁盘;由于扇区比较小,数量非常多,

在寻址比较麻烦,操作系统就将相邻的几个扇区组合在一起,形成簇,再以簇为每次作文件的最小单元。比如加载一个磁盘文件内容,

操作系统是分批次读取,每次只拷贝一个簇的单位数据,我的电脑就是一次拷贝4096字节,知道文件全部拷贝完成。

2)读写速度

  磁盘读取时间是毫秒级别的一般几毫秒到十几毫秒之间,这个跟磁盘转速有点关系,还有就是数据所在磁道远近有关系;

CPU处理时间是纳秒级别,毫秒:纳秒 = 1:1000000,所以在程序设计中,读取文件是时间成本非常高的,应该尽量合理设计;

 

B树简介(维基百科):

  B树(英语:B-tree)是一种自平衡的树,能够保持数据有序。这种数据结构能够让查找数据、顺序访问、插入数据及删除的动作,

都在对数时间内完成。B树,概括来说是一个一般化的二叉查找树(binary search tree)一个节点可以拥有最少2个子节点。

与自平衡二叉查找树不同,B树适用于读写相对大的数据块的存储系统,例如磁盘。B树减少定位记录时所经历的中间过程,从而加快存取速度。

B树这种数据结构可以用来描述外部存储。这种数据结构常被应用在数据库和文件系统的实现上。

  一个 m 阶的B树是一个有以下特性:

  • 每一个节点最多有 m 个子节点
  • 每一个非叶子节点(除根节点)最少有 ⌈m/2⌉ 个子节点
  • 如果根节点不是叶子节点,那么它至少有两个子节点
  • 有 k 个子节点的非叶子节点拥有 k − 1 个键
  • 所有的叶子节点都在同一层

 

  好吧,上边这一段看了等于没看的定义可以不看,这里有个重要的B树特性需要了解,就是B树的阶,对于阶的定义国内外是有分歧的,有的定义为度

阶指的是节点的最大孩子数,度指的是节点的最小孩子数,我查阅了很多资料,基本上可以理解为: 

1度 = 2阶,比如说3度B树,可以理解为6阶B树。这点有些疑问,有更好的说法的可以留言讨论一下。

 

1)内部节点:

  内部节点是除叶子节点和根节点之外的所有节点。每个内部节点拥有最多 U 个,最少 L 个子节点。元素的数量总是比子节点指针的数量少1。

U 必须等于 2L 或者 2L-1。这个L一般是度数。

2)根节点:根节点拥有的子节点数量的上限和内部节点相同,但是没有下限。

3)叶子节点:叶子节点对元素的数量有相同的限制,但是没有子节点,也没有指向子节点的指针。

4)为了分析方便举例3阶3层B树

 

                        图1 

从上图中可以得出以下几个信息:

  • 红色数字标示整个节点(即3、6在同一个节点内,图中总共9个节点),黑色数字表示每个节点内的键值。
  • 所有数据插入B树后,都是从左到右顺序排列,从根节点开始,节点左边孩子键值都小于节点键值,右边孩子键值都大于节点键值。
  • 树的阶数指的是每个节点的最大孩子节点数,图中最多孩子节点数为3,即阶数=3,键值数量最少为:1,最大为:阶数 -1

 

数据检索分析:

  依据上图分析,因为整棵树已经在内存中,相当于一个变量,数据检索首先是从根节点开始;

1)如果要查询9,首先从根节点比较,那比较一次就得到结果,

2)如果要查询第二层的3、4,首先判断根节点键值,没有匹配到,但是可以判断要检索的键值比根节点小,

   所以接下来是从左孩子树继续检索,12、15也是类似,总共需要2次比较就得到结果

3)如果查询叶子节点键值,类似2),只需要3次比较就能得到结果。

4)对比普通的数组遍历查询,B树检索的时间成本没有随数据量增加而线性增加,效率大大提高。

 

B树的应用分析:

  前面已经提到,如果树已经在内存中,那当然好办,直接遍历就好了。如果B树仅仅如此,那也和数组差别不大,同样受限于内存大小;

所以,在内存中创建整棵B树是不现实的,这不是B树的正确打开方式。

  前面也已经提到,操作系统加载磁盘文件的时候,如果文件超过大小(即4096个字节),那会分多次的读取磁盘,直到拷贝数据完成。

这里看似一个加载动作,其实这个动作包含了N次磁盘寻址,而我们已经知道,每次磁盘寻址直至拷贝数据开销是非常大的;是CPU指令耗时百万倍以上;

这种操作应该尽量少地执行,而B树这种数据结构就是为了解决磁盘读取瓶颈这个问题而产生的。

  实际应用中,B树会持久化到磁盘,然后只在内存保留一个根节点的指针。已上图1为例:

  每个节点大小刚好等于大小,这样只需一次磁盘IO就可以获取到一整个节点的所有键值,及其所有子树的指针。

比如,查询键值8:

  1)第一步,读取根节点得到键值9,以及2个子树指针,分别指向左右孩子节点,因为9 > 8,所以下一步加载左孩子节点

  2)第二部,加载节点2,得到键值3、6,以及3个子树指针,因为3、6 < 8,所以下一步要加载节点2的右孩子节点

  3)第三部,加载节点6,得到键值7、8,因为是叶子节点所以没有子树指针,遍历键值匹配到8,返回。

 

总结:

  在这个3阶3层的B树中,无论查找哪一个键值,最多只需要3次磁盘操作,就算平均每次耗时10毫秒,总共需要耗时30毫秒(CPU运算耗时可以忽略);

以此类推,3阶4层的B树,需要读取4次磁盘,耗时40毫秒,5层50毫秒,6层60毫秒,7层,8层,,,,

  这样一看貌似也没什么,几十毫秒已经不能说快了,但是别忘了我们这颗树只有3阶,即一个节点保存2个键值。一个簇最多能有4096/4=1024个键值;

如果创建一个1024阶的B树,分别控制在3、4、5层的话,根据B树高度公式:,H为层数,T为1024,n为数据总数

耗时如下:

  3阶3层:能容纳2147483648(20亿)个键值,检索耗时也将30毫秒内

  3阶4层:能容纳2147483648(20亿) ~ 2199023255552(2兆亿)个键值,检索耗时也将40毫秒内,当然这已经超出键值表达范围了

  3阶5层:不可思议。。。

 

  当然实际运用当中达不到1024阶,因为树持久化到磁盘时,索引结构体一般都是超过4个字节,比如12个字节,那一个簇最多能有4096/12=341个键值。

如果阶数按341来算:

  3阶3层:能容纳79303642(7千万)个键值,检索耗时也将30毫秒内

  3阶4层:能容纳79303642(7千万) ~ 27042541922(200亿)个键值,检索耗时也将40毫秒内

  也是非常多了。。

 

B树简单示例:

1)首先,我们把B树基本信息定义出来

1 public class Consts
2 {
3     public const int M = 3;                  // B树的最小度数
4     public const int KeyMax = 2 * M - 1;     // 节点包含关键字的最大个数
5     public const int KeyMin = M - 1;         // 非根节点包含关键字的最小个数
6     public const int ChildMax = KeyMax + 1;  // 孩子节点的最大个数
7     public const int ChildMin = KeyMin + 1;  // 孩子节点的最小个数
8 }

先写个简单的demo,因为最小度数为3,那就是6阶。先实现几个简单的方法,新增,拆分,其余的合并,删除比较复杂以后有机会再看看

2)定义BTreeNode,B树节点


  1     public class BTreeNode
  2     {
  3         private bool leaf;
  4         public int[] keys;
  5         public int keyNumber;
  6         public BTreeNode[] children;
  7         public int blockIndex;
  8         public int dataIndex;
  9 
 10         public BTreeNode(bool leaf)
 11         {
 12             this.leaf = leaf;
 13             keys = new int[Consts.KeyMax];
 14             children = new BTreeNode[Consts.ChildMax];
 15         }
 16 
 17         /// <summary>在未满的节点中插入键值</summary>
 18         /// <param name="key">键值</param>
 19         public void InsertNonFull(int key)
 20         {
 21             var index = keyNumber - 1;
 22 
 23             if (leaf == true)
 24             {
 25                 // 找到合适位置,并且移动节点键值腾出位置
 26                 while (index >= 0 && keys[index] > key)
 27                 {
 28                     keys[index + 1] = keys[index];
 29                     index--;
 30                 }
 31 
 32                 // 在index后边新增键值
 33                 keys[index + 1] = key;
 34                 keyNumber = keyNumber + 1;
 35             }
 36             else
 37             {
 38                 // 找到合适的子孩子索引
 39                 while (index >= 0 && keys[index] > key) index--;
 40 
 41                 // 如果孩子节点已满
 42                 if (children[index + 1].keyNumber == Consts.KeyMax)
 43                 {
 44                     // 分裂该孩子节点
 45                     SplitChild(index + 1, children[index + 1]);
 46 
 47                     // 分裂后中间节点上跳父节点
 48                     // 孩子节点已经分裂成2个节点,找到合适的一个
 49                     if (keys[index + 1] < key) index++;
 50                 }
 51 
 52                 // 插入键值
 53                 children[index + 1].InsertNonFull(key);
 54             }
 55         }
 56 
 57         /// <summary>分裂节点</summary>
 58         /// <param name="childIndex">孩子节点索引</param>
 59         /// <param name="waitSplitNode">待分裂节点</param>
 60         public void SplitChild(int childIndex, BTreeNode waitSplitNode)
 61         {
 62             var newNode = new BTreeNode(waitSplitNode.leaf);
 63             newNode.keyNumber = Consts.KeyMin;
 64 
 65             // 把待分裂的节点中的一般节点搬到新节点
 66             for (var j = 0; j < Consts.KeyMin; j++)
 67             {
 68                 newNode.keys[j] = waitSplitNode.keys[j + Consts.ChildMin];
 69 
 70                 // 清0
 71                 waitSplitNode.keys[j + Consts.ChildMin] = 0;
 72             }
 73 
 74             // 如果待分裂节点不是也只节点
 75             if (waitSplitNode.leaf == false)
 76             {
 77                 for (var j = 0; j < Consts.ChildMin; j++)
 78                 {
 79                     // 把孩子节点也搬过去
 80                     newNode.children[j] = waitSplitNode.children[j + Consts.ChildMin];
 81 
 82                     // 清0
 83                     waitSplitNode.children[j + Consts.ChildMin] = null;
 84                 }
 85             }
 86 
 87             waitSplitNode.keyNumber = Consts.KeyMin;
 88 
 89             // 拷贝一般键值到新节点
 90             for (var j = keyNumber; j >= childIndex + 1; j--)
 91                 children[j + 1] = children[j];
 92 
 93             children[childIndex + 1] = newNode;
 94             for (var j = keyNumber - 1; j >= childIndex; j--)
 95                 keys[j + 1] = keys[j];
 96 
 97             // 把中间键值上跳至父节点
 98             keys[childIndex] = waitSplitNode.keys[Consts.KeyMin];
 99 
100             // 清0
101             waitSplitNode.keys[Consts.KeyMin] = 0;
102 
103             // 根节点键值数自加
104             keyNumber = keyNumber + 1;
105         }
106 
107         /// <summary>根据节点索引顺序打印节点键值</summary>
108         public void PrintByIndex()
109         {
110             int index;
111             for (index = 0; index < keyNumber; index++)
112             {
113                 // 如果不是叶子节点, 先打印叶子子节点. 
114                 if (leaf == false) children[index].PrintByIndex();
115 
116                 Console.Write("{0} ", keys[index]);
117             }
118 
119             // 打印孩子节点
120             if (leaf == false) children[index].PrintByIndex();
121         }
122 
123         /// <summary>查找某键值是否已经存在树中</summary>
124         /// <param name="key">键值</param>
125         /// <returns></returns>
126         public BTreeNode Find(int key)
127         {
128             int index = 0;
129             while (index < keyNumber && key > keys[index]) index++;
130 
131             // 该key已经存在, 返回该索引位置节点
132             if (keys[index] == key) return this;
133 
134             // key 不存在,并且节点是叶子节点
135             if (leaf == true) return null;
136 
137             // 递归在孩子节点中查找
138             return children[index].Find(key);
139         }
140     }

View Code

3)B树模型


 1     public class BTree
 2     {
 3         public BTreeNode Root { get; private set; }
 4 
 5         public BTree() { }
 6 
 7         /// <summary>根据节点索引顺序打印节点键值</summary>
 8         public void PrintByIndex()
 9         {
10             if (Root == null)
11             {
12                 Console.WriteLine("空树");
13                 return;
14             }
15 
16             Root.PrintByIndex();
17         }
18 
19         /// <summary>查找某键值是否已经存在树中</summary>
20         /// <param name="key">键值</param>
21         /// <returns></returns>
22         public BTreeNode Find(int key)
23         {
24             if (Root == null) return null;
25 
26             return Root.Find(key);
27         }
28 
29         /// <summary>新增B树节点键值</summary>
30         /// <param name="key">键值</param>
31         public void Insert(int key)
32         {
33             if (Root == null)
34             {
35                 Root = new BTreeNode(true);
36                 Root.keys[0] = key; 
37                 Root.keyNumber = 1;  
38                 return;
39             }
40 
41             if (Root.keyNumber == Consts.KeyMax)
42             {
43                 var newNode = new BTreeNode(false);
44 
45                 newNode.children[0] = Root; 
46                 newNode.SplitChild(0, Root);
47 
48                 var index = 0;
49                 if (newNode.keys[0] < key) index++;
50 
51                 newNode.children[index].InsertNonFull(key);
52                 Root = newNode;
53             }
54             else
55             {
56                 Root.InsertNonFull(key);
57             }
58         }
59     }

View Code

4)新增20个无序键值,测试一下

 1             var bTree = new BTree();
 2 
 3             bTree.Insert(4);
 4             bTree.Insert(5);
 5             bTree.Insert(6);
 6             bTree.Insert(1);
 7             bTree.Insert(2);
 8             bTree.Insert(3);
 9             bTree.Insert(10);
10             bTree.Insert(11);
11             bTree.Insert(12);
12             bTree.Insert(7);
13             bTree.Insert(8);
14             bTree.Insert(9);
15             bTree.Insert(13);
16             bTree.Insert(14);
17             bTree.Insert(18);
18             bTree.Insert(19);
19             bTree.Insert(20);
20             bTree.Insert(15);
21             bTree.Insert(16);
22             bTree.Insert(17);
23 
24             Console.WriteLine("输出排序后键值");
25             bTree.PrintByIndex();

5)运行

 

B树持久化: 

  上文提到,B数不可能只存在内存而无法落地,那样没有意义。所以就需要将整棵树持久化到磁盘文件,并且还要支持快速地从磁盘文件中检索到键值;

要持久化就要考虑很多问题,像上边的简单示例是没有实际意义的,因为节点不可能只有键值与孩子树,还得有数据指针,存储位置等等,大概有以下一些问题:

  • 如何保存每个节点占有字节数刚好等于一个簇大小(4096字节),因为这样就符合一次IO操作的数据交换上限?
  • 如何保存每个节点的所有键值,以及这个节点下属所有子树关系?
  • 如何保存每个键值对应的数据指针地址,以及指针与键值的对应关系如何维持?
  • 如何保证内存与磁盘的数据交换中能够正确地还原树结构,即重建树的某部分层级与键值和子树的关系?
  • 等等。。

  问题比较多,非常麻烦。具体的过程就不列举了,以下展示以下修改后的B树模型。

 

1、先定义一个结构体

 1 [StructLayout(LayoutKind.Sequential, CharSet = CharSet.Ansi, Pack = 1)]
 2 public struct BlockItem
 3 {
 4     public int ChildBlockIndex;
 5     public int Key;
 6     public int DataIndex;
 7 
 8     public BlockItem(int key, int dataIndex)
 9     {
10         ChildBlockIndex = -1;
11         Key = key;
12         DataIndex = dataIndex;
13     }
14 }

 

  结构体总共12字节,为了能够持久化整棵B树到磁盘,加入了ChildBlockIndex子孩子节点块索引,根据这个块索引在下一次重建子孩子树层级关系时就知道从

文件的那个位置开始读取;Key键值,DataIndex数据索引,数据索引也是一个文件位置记录,跟ChildBlockIndex差不多,这样检索到key后就知道从

文件哪个位置获取真正的数据。为了更形象了解B树应用,我画了一个结构体的示意图:

0、总共3个节点,每个节点由N个结构体组成,最末尾只有孩子指针,没有数据与键值

1、黄色为子树块索引,即ChildBlockIndex,指向这个子孩子树所有数据在文件中的位置

2、红色为键值,即Key,键值一般是唯一的,不允许重复

3、蓝色为数据块索引,即DataIndex,指向键值对应的数据在文件中的什么位置开始,然后读取一个结构体的长度即可

4、底下绿色的一块是数据指针指向的具体数据块

2、数据结构体

 1 [StructLayout(LayoutKind.Sequential, CharSet = CharSet.Ansi, Pack = 1)]
 2 public struct SDataTest
 3 {
 4     public int Idx;
 5     public int Age;
 6     public byte Sex;
 7 
 8     [MarshalAs(UnmanagedType.ByValArray, SizeConst = 20)]
 9     public byte[] Name;
10 
11     public byte Valid;
12 };

 

3、B树节点类修改改一下,这个就不解释了,复习一下程序员基本功,啃代码。


  1     public class BTreeNode
  2     {
  3         private BTree tree;
  4         private bool leaf;
  5 
  6         public int keyNumber;
  7         public BlockItem[] keys;
  8         public BTreeNode[] children;
  9 
 10         public int blockIndex;
 11         public int findIndex;
 12 
 13         public BTreeNode(BTree tree, bool leaf)
 14         {
 15             this.tree = tree;
 16             this.leaf = leaf;
 17             keys = new BlockItem[Consts.KeyMax];
 18             children = new BTreeNode[Consts.ChildMax];
 19             blockIndex = Consts.BlockIndex++;
 20         }
 21 
 22         /// <summary>在未满的节点中插入键值</summary>
 23         /// <param name="key">键值</param>
 24         public void InsertNonFull(BlockItem item)
 25         {
 26             var index = keyNumber - 1;
 27 
 28             if (leaf == true)
 29             {
 30                 // 找到合适位置,并且移动节点键值腾出位置
 31                 while (index >= 0 && keys[index].Key > item.Key)
 32                 {
 33                     keys[index + 1] = keys[index];
 34                     index--;
 35                 }
 36 
 37                 // 在index后边新增键值
 38                 keys[index + 1] = item;
 39                 keyNumber = keyNumber + 1;
 40             }
 41             else
 42             {
 43                 // 找到合适的子孩子索引
 44                 while (index >= 0 && keys[index].Key > item.Key) index--;
 45 
 46                 // 如果孩子节点已满
 47                 if (children[index + 1].keyNumber == Consts.KeyMax)
 48                 {
 49                     // 分裂该孩子节点
 50                     SplitChild(index + 1, children[index + 1]);
 51 
 52                     // 分裂后中间节点上跳父节点
 53                     // 孩子节点已经分裂成2个节点,找到合适的一个
 54                     if (keys[index + 1].Key < item.Key) index++;
 55                 }
 56 
 57                 // 插入键值
 58                 children[index + 1].InsertNonFull(item);
 59             }
 60         }
 61 
 62         /// <summary>分裂节点</summary>
 63         /// <param name="childIndex">孩子节点索引</param>
 64         /// <param name="waitSplitNode">待分裂节点</param>
 65         public void SplitChild(int childIndex, BTreeNode waitSplitNode)
 66         {
 67             var newNode = new BTreeNode(tree, waitSplitNode.leaf);
 68             newNode.keyNumber = Consts.KeyMin;
 69 
 70             // 把待分裂的节点中的一般节点搬到新节点
 71             for (var j = 0; j < Consts.KeyMin; j++)
 72             {
 73                 newNode.keys[j] = waitSplitNode.keys[j + Consts.ChildMin];
 74 
 75                 // 清0
 76                 waitSplitNode.keys[j + Consts.ChildMin] = default(BlockItem);
 77             }
 78 
 79             // 如果待分裂节点不是也只节点
 80             if (waitSplitNode.leaf == false)
 81             {
 82                 for (var j = 0; j < Consts.ChildMin; j++)
 83                 {
 84                     // 把孩子节点也搬过去
 85                     newNode.children[j] = waitSplitNode.children[j + Consts.ChildMin];
 86 
 87                     // 清0
 88                     waitSplitNode.children[j + Consts.ChildMin] = null;
 89                 }
 90             }
 91 
 92             waitSplitNode.keyNumber = Consts.KeyMin;
 93 
 94             for (var j = keyNumber; j >= childIndex + 1; j--)
 95                 children[j + 1] = children[j];
 96 
 97             children[childIndex + 1] = newNode;
 98 
 99             for (var j = keyNumber - 1; j >= childIndex; j--)
100                 keys[j + 1] = keys[j];
101 
102             // 把中间键值上跳至父节点
103             keys[childIndex] = waitSplitNode.keys[Consts.KeyMin];
104 
105             // 清0
106             waitSplitNode.keys[Consts.KeyMin] = default(BlockItem);
107 
108             // 根节点键值数自加
109             keyNumber = keyNumber + 1;
110         }
111 
112         /// <summary>根据节点索引顺序打印节点键值</summary>
113         public void PrintByIndex()
114         {
115             int index;
116             for (index = 0; index < keyNumber; index++)
117             {
118                 // 如果不是叶子节点, 先打印叶子子节点. 
119                 if (leaf == false) children[index].PrintByIndex();
120 
121                 Console.Write("{0} ", keys[index].Key);
122             }
123 
124             // 打印孩子节点
125             if (leaf == false) children[index].PrintByIndex();
126         }
127 
128         /// <summary>查找某键值是否已经存在树中</summary>
129         /// <param name="item">键值</param>
130         /// <returns></returns>
131         public BTreeNode Find(BlockItem item)
132         {
133             findIndex = 0;
134             int index = 0;
135             while (index < keyNumber && item.Key > keys[index].Key) index++;
136 
137             // 遍历全部都未找到,索引计数减1
138             if (index > 0 && index == keyNumber) index--;
139 
140             // 该key已经存在, 返回该索引位置节点
141             if (keys[index].Key == item.Key)
142             {
143                 findIndex = index;
144                 return this;
145             }
146 
147             // key 不存在,并且节点是叶子节点
148             if (leaf == true) return null;
149 
150             // 重建children[index]数据结构
151             var childBlockIndex = keys[index].ChildBlockIndex;
152             tree.LoadNodeByBlock(ref children[index], childBlockIndex);
153 
154             // 递归在孩子节点中查找
155             if (children[index] == null) return null;
156             return children[index].Find(item);
157         }
158     }

View Code

 

4、B树模型也要修改一下 ,不解释


  1     public class BTree
  2     {
  3         private FileStream rwFS;
  4 
  5         public BTreeNode Root;
  6 
  7         public BTree(string fullName)
  8         {
  9             rwFS = new FileStream(fullName, FileMode.OpenOrCreate, FileAccess.ReadWrite);
 10 
 11             // 创建10M的空间,用做索引存储
 12             if (rwFS.Length == 0)
 13             {
 14                 rwFS.SetLength(Consts.IndexTotalSize);
 15             }
 16 
 17             // 从数据文件重建根节点,内存只保存根节点
 18             LoadNodeByBlock(ref Root, 0);
 19         }
 20 
 21         public void LoadNodeByBlock(ref BTreeNode node, int blockIndex)
 22         {
 23             var items = Helper.Read(rwFS,blockIndex);
 24             if (items.Count > 0)
 25             {
 26                 var isLeaf = items[0].ChildBlockIndex == Consts.NoChild;
 27 
 28                 node = new BTreeNode(this, isLeaf);
 29                 node.blockIndex = blockIndex;
 30                 node.keys = items.ToArray();
 31                 node.keyNumber = items.Count;
 32             }
 33         }
 34 
 35         /// <summary>根据节点索引顺序打印节点键值</summary>
 36         public void PrintByIndex()
 37         {
 38             if (Root == null)
 39             {
 40                 Console.WriteLine("空树");
 41                 return;
 42             }
 43 
 44             Root.PrintByIndex();
 45         }
 46 
 47         /// <summary>查找某键值是否已经存在树中</summary>
 48         /// <param name="item">键值</param>
 49         /// <returns></returns>
 50         public BTreeNode Find(BlockItem item)
 51         {
 52             if (Root == null) return null;
 53 
 54             return Root.Find(item);
 55         }
 56         public BTreeNode Find(int key)
 57         {
 58             return Find(new BlockItem() { Key = key });
 59         }
 60 
 61         /// <summary>新增B树节点键值</summary>
 62         /// <param name="item">键值</param>
 63         private void Insert(BlockItem item)
 64         {
 65             if (Root == null)
 66             {
 67                 Root = new BTreeNode(this, true);
 68                 Root.keys[0] = item;  
 69                 Root.keyNumber = 1;  
 70             }
 71             else
 72             {
 73                 if (Root.keyNumber == Consts.KeyMax)
 74                 {
 75                     var newNode = new BTreeNode(this, false);
 76 
 77                     newNode.children[0] = Root;
 78                     newNode.SplitChild(0, Root);
 79 
 80                     var index = 0;
 81                     if (newNode.keys[0].Key < item.Key) index++;
 82 
 83                     newNode.children[index].InsertNonFull(item);
 84                     Root = newNode;
 85                 }
 86                 else
 87                 {
 88                     Root.InsertNonFull(item);
 89                 }
 90             }
 91         }
 92 
 93         public void Insert(SDataTest data)
 94         {
 95             var item = new BlockItem()
 96             {
 97                 Key = data.Idx
 98             };
 99 
100             var node = Find(item);
101             if (node != null)
102             {
103                 Console.WriteLine("键值已经存在,info:{0}", item.Key);
104                 return;
105             }
106 
107             // 保存数据
108             item.DataIndex = Helper.InsertData(rwFS, data);
109 
110             // 保存索引
111             if (item.DataIndex >= 0)
112                 Insert(item);
113         }
114 
115         /// <summary>持久化整棵树</summary>
116         public void SaveIndexAll()
117         {
118             SaveIndex(Root);
119         }
120 
121         /// <summary>持久化某节点以下的树枝</summary>
122         /// <param name="node">某节点</param>
123         public void SaveIndex(BTreeNode node)
124         {
125             var bw = new BinaryWriter(rwFS);
126             var keyItem = default(BlockItem);
127 
128             // 第一层
129             var nodeL1 = node;
130             if (nodeL1 == null) return;
131 
132             for (var i = 0; i <= nodeL1.keyNumber; i++)
133             {
134                 keyItem = default(BlockItem);
135                 if (i < nodeL1.keyNumber) keyItem = nodeL1.keys[i];
136 
137                 SaveIndex(bw, 0, i, nodeL1.children[i], keyItem);
138 
139                 // 第二层
140                 var nodeL2 = nodeL1.children[i];
141                 if (nodeL2 == null) continue;
142 
143                 for (var j = 0; j <= nodeL2.keyNumber; j++)
144                 {
145                     keyItem = default(BlockItem);
146                     if (j < nodeL2.keyNumber) keyItem = nodeL2.keys[j];
147 
148                     SaveIndex(bw, nodeL2.blockIndex, j, nodeL2.children[j], keyItem);
149 
150                     // 第三层
151                     var nodeL3 = nodeL2.children[j];
152                     if (nodeL3 == null) continue;
153 
154                     for (var k = 0; k <= nodeL3.keyNumber; k++)
155                     {
156                         keyItem = default(BlockItem);
157                         if (k < nodeL3.keyNumber) keyItem = nodeL3.keys[k];
158 
159                         SaveIndex(bw, nodeL3.blockIndex, k, nodeL3.children[k], keyItem);
160 
161                         // 第四层
162                         var nodeL4 = nodeL3.children[k];
163                         if (nodeL4 == null) continue;
164 
165                         for (var l = 0; l <= nodeL4.keyNumber; l++)
166                         {
167                             keyItem = default(BlockItem);
168                             if (l < nodeL4.keyNumber) keyItem = nodeL4.keys[l];
169 
170                             SaveIndex(bw, nodeL4.blockIndex, l, nodeL4.children[l], keyItem);
171 
172                             // 第五层
173                             var nodeL5 = nodeL4.children[l];
174                             if (nodeL5 == null) continue;
175 
176                             for (var z = 0; z <= nodeL5.keyNumber; z++)
177                             {
178                                 keyItem = default(BlockItem);
179                                 if (z < nodeL5.keyNumber) keyItem = nodeL5.keys[z];
180 
181                                 SaveIndex(bw, nodeL5.blockIndex, z, nodeL5.children[z], keyItem);
182                             }
183                         }
184                     }
185                 }
186             }
187         }
188         private void SaveIndex(BinaryWriter bw, int blockIndex, int num, BTreeNode node, BlockItem item)
189         {
190             bw.Seek((blockIndex * Consts.BlockSize) + (num * Consts.IndexSize), SeekOrigin.Begin);
191             bw.Write(node == null ? Consts.NoChild : node.blockIndex);
192             bw.Write(item.Key);
193             bw.Write(item.DataIndex);
194             bw.Flush();
195         }
196 
197         public SDataTest LoadData(int dataIndex)
198         {
199             return Helper.Load(rwFS, dataIndex);
200         }
201     }

View Code

 

5、写测试

 1 private static void InsertTest(ref BTree bTree)
 2 {
 3     // 新增测试数据
 4     for (int i = 1; i <= Consts.TotalKeyNumber; i++)
 5     {
 6         bTree.Insert(new SDataTest()
 7         {
 8             Idx = i,
 9             Age = i,
10             Sex = 1,
11             Name = Helper.Copy("Name(" + i.ToString() + ")", 20),
12             Valid = 1
13         });
14     }
15 
16     Console.WriteLine("测试数据添加完毕,共新增{0}条数据", Consts.TotalKeyNumber);
17 }

 

6、读测试

 1 private static void FindTest(ref BTree bTree)
 2 {
 3     var count = 0;
 4 
 5     // 校验数据查找
 6     for (int i = 1; i <= Consts.TotalKeyNumber; i++)
 7     {
 8         var node = bTree.Find(i);
 9         if (node == null)
10         {
11             //Console.WriteLine("未找到{0}", i);
12             continue;
13         }
14 
15         //Console.WriteLine("findIndex:{0},key:{1},dataIndex:{2}", node.findIndex, node.keys[node.findIndex].Key, node.keys[node.findIndex].DataIndex);
16 
17         count++;
18         if (count % 10000 == 0)
19         {
20             var data = bTree.LoadData(node.keys[node.findIndex].DataIndex);
21             var name = Encoding.Default.GetString(data.Name).TrimEnd('\0');
22             Console.WriteLine("Idx:{0},Age:{1},Sex:{2},Name:{3},Valid:{4}", data.Idx, data.Age, data.Sex, name, data.Valid);
23         }
24     }
25 
26     Console.WriteLine("有效数据个数:{0}", count);
27 }

 

7、最后测试一下

 

 8、测试查询时间

 1 private static void CheckLoadTime(ref BTree bTree, int key)
 2 {
 3     var start = DateTime.Now;
 4     var node = bTree.Find(key);
 5     if (node == null) return;
 6 
 7     Console.WriteLine("查找{0},耗时:{1}", key.ToString(), (DateTime.Now - start).TotalMilliseconds.ToString());
 8 
 9     var data = bTree.LoadData(node.keys[node.findIndex].DataIndex);
10     var name = Encoding.Default.GetString(data.Name).TrimEnd('\0');
11     Console.WriteLine("Idx:{0},Age:{1},Sex:{2},Name:{3},Valid:{4}", data.Idx, data.Age, data.Sex, name, data.Valid);
12     Console.WriteLine();
13 }
1      CheckLoadTime(ref bTree, 1000);
2      CheckLoadTime(ref bTree, 10000);
3      CheckLoadTime(ref bTree, 50000);
4      CheckLoadTime(ref bTree, 100000);

 

 9、重新生成10000000条数据,测试查询效率

1      CheckLoadTime(ref bTree, 100000);
2      CheckLoadTime(ref bTree, 1000000);
3      CheckLoadTime(ref bTree, 3000000);
4      CheckLoadTime(ref bTree, 5000000);
5      CheckLoadTime(ref bTree, 8000000);
6      CheckLoadTime(ref bTree, 10000000);

 

全是1毫秒内返回,数据检索效率非常高,

 

学习历程:

  实际上最初在学校潦草学了一遍【数据结构】之后,工作那么多年都用不着这方面的知识点,早就忘得一干二净了。

重新引起我兴趣的是2017年下半年,当时一个项目需要用到共享内存作为快速读写数据的底层核心功能。在设计共享内存存储关系时,

就遇到了索引的快速检索要求,第一次是顺序检索,当数据量达到5万以上时系统就崩了,检索速度太慢;后来改为二分查找法,轻松达到20万数据;

达到20万后就差不多到了单机处理性能瓶颈了,因为CPU不够用,除了检索还需要做其他的业务计算;

  那时候就一直在搜索快速查找的各种算法,什么快速排序算法、堆排序算法、归并排序、二分查找算法、DFS(深度优先搜索)、BFS(广度优先搜索),

基本上都了解了一遍,但是看得头疼,没去实践。最后看到树结构,引起我很大兴趣,就是园友nullzx的这篇:B+树在磁盘存储中的应用

这让我了解到原来数据库是这样读写的,这很有意思,得造个轮子自己试一次。最后性能优化的事丢一边,学习B树去了,哈(掩面笑.jpg)。

      学习毫无疑问是枯燥反人性的,我基本就是看一下,玩一下,一直断断续续的;上下班地铁上

刷一下B树的介绍与运用。周末窝在宿舍,看着电脑里的算法导论就想睡觉;周末去旁边的铁仔山,山崖地下坐着无事,又看了看磁盘的构造与工作原理。在铁仔山顶,远眺着远处的珠江入海口,时不时地想着B树的持久化方法;

        上班,生活,忙碌又有点平淡。整天做码农也不好,总得找点有趣的事情做做是吧。

  粗陋仓促写成,恐怕有很多地方有漏洞,所以如果文中有错误的地方,欢迎留言讨论,但是拒绝一波流的吐槽,我可是会删低级评论的。

 

赞(0) 打赏
未经允许不得转载:开源SEO开源SEO博客 » B树概述与简单应用示例(C#)

开源SEO 更专业 更方便

联系我们投稿网址