使用php实现一个简单的key-value数据库

最近在阅读《php核心技术与最佳实践》，前面的部分只是大体了解了下。不过当读到Hash算法与数据库实现这一章时，文中给出了一个使用php实现一个简单的key-value数据库的实例，这引起了我的兴趣。仔细阅读后觉得不过瘾，所以仿照给出的例子自己实践了一遍，来巩固一下学习的内容，同时加深下对数据的理解。

1.文中实现了Hash索引，Hash值的冲突，通过分离链表法实现。

2.实现的索引类型是非聚簇索引，即索引文件和数据文件分离(示例中分别保存到index和data文件夹)

3.Hash表中链表单个节点的数据结构是：下一个索引的位置(4) + keyValue(128) + 数据文件中的位置(4) + 数据长度(4);

4.删除操作仅仅是删除了索引，在数据文件中此key对应的值还是存在的。如果在删除操作时同时删除数据文件中的数据，这需要对索引文件中的所有节点的数据指针刷新，操作耗时过大。

下面是实现代码：

一.数据库连接部分

<?php

/*
*
*   索引文件使用pack和unpack函数，将索引文件数据转为二进制和从二进制转到具体值
*
*   在当前文件夹下创建data和index文件夹，赋予它可读写权限，简单点直接赋予777权限。
*
*   $db = new MyDataBase();
*
*    //数据库连接
*   $dbHandler = $db->connect('dbTest');
*
*   //写入数据，仅支持key-value形式
*   $dbHandler->insert('key1', '1111111');
*
*   //数据查找
*   $dbHandler->find('key1');
*
*   //删除数据
*   $dbHandler->delete('key1');
*
*   //关闭数据连接
*   $db->close();
*
*/
define('DB_INSERT_SUCCESS', 'SUCCESS');
define('DB_INSERT_FAILED', 'FAILED');
define('DB_DELETE_SUCCESS', 'SUCCESS');
define('DB_DELETET_FAILED', 'FAILED');
define('DB_EXISTS_KEY', 'KEYEXISTS');

//这里定义的是Hash表中链表的长度
define('DB_BUCKET_SIZE', 262144);

//存储key值的大小
define('DB_KEY_SIZE', 128);

//定义了一个索引节点的长度，一个索引节点包括:
//4字节的指向下一索引节点值 + 128字节的Key值 +
//4字节的数据在数据文件的偏移值 + 4字节的数据长度值
define('DB_INDEX_SIZE', DB_KEY_SIZE + 12);

include './DataBaseObject.php';

class MyDataBase
{
private $dataHandler;
private $indexHandler;
private $closeFlag = false;

public function connect($databaseName)
{
try {

$indexFile = './index/'.$databaseName.'.inx';
$dataFile = './data/'.$databaseName.'.dat';

$needInit = false;

if (!file_exists($indexFile)) {
$openModel = 'w+b';
$needInit = true;
} else {
$openModel = 'r+b';
}

$this->indexHandler = fopen($indexFile, $openModel);

//初始化索引文件，用0初始化。
if ($needInit) {
$initValue = pack('L', 0x00000000);
for ($i = 0; $i < DB_BUCKET_SIZE; $i++) {
fwrite($this->indexHandler, $initValue, 4);
}
}

$this->dataHandler = fopen($dataFile, $openModel);

$dataBase = new DataBaseObject($this->dataHandler, $this->indexHandler);

} catch (Exception $e) {
return NULL;
}
return $dataBase;
}

public function close()
{
try {
if (!$this->closeFlag) {
fclose($this->dataHandler);
fclose($this->indexHandler);
$this->closeFlag = true;
}
} catch (Exception $e) {
return false;
}
return true;
}
}

//下面是使用示例
$db = new MyDataBase();

$dbHandler = $db->connect('dbTest');

$dbHandler->insert('key1', '1111111');
var_dump($dbHandler->find('key1'));

$dbHandler->delete('key1');
var_dump($dbHandler->find('key1'));

$db->close();

二.数据表操作部分

<?php

/*
*
*   使用的hash函数是hashFunc，先使用md5转为32位，
*   再将得到的32位字符串的前8位使用Times33求得hash值。
*
*/
class DataBaseObject
{
private $dataHandler;
private $indexHandler;

public function __construct($dataHandler, $indexHandler)
{
$this->dataHandler = $dataHandler;
$this->indexHandler = $indexHandler;
}

public function insert($key, $data)
{
$offset = $this->hashFunc($key) % DB_BUCKET_SIZE * 4;
//获取下一个可用的磁盘地址的偏移量
$indexOffset = fstat($this->indexHandler);
$indexOffset = intval($indexOffset['size']);

$dataOffset = fstat($this->dataHandler);
$dataOffset = intval($dataOffset['size']);

$keyLen = strlen($key);

if ($keyLen > DB_KEY_SIZE) {
return DB_INSERT_FAILED;
}

//新节点指向的下一个节点是0
$dataBlock = pack('L', 0x00000000);
$dataBlock .= $key;
$space = DB_KEY_SIZE - $keyLen;
for($i = 0; $i < $space; $i++) {
//长度不够，用0补齐。
$dataBlock .= pack('C', 0x00);
}

//新数据在数据文件中的偏移量
$dataBlock .= pack('L', $dataOffset);
//新数据长度
$dataBlock .= pack('L', strlen($data));

fseek($this->indexHandler, $offset, SEEK_SET);
$position = unpack('L', fread($this->indexHandler, 4));
$position = $position[1];

//如果该散列值从未出现过，直接作为头节点
if ($position == 0) {
fseek($this->indexHandler, $offset, SEEK_SET);
//头节点指向当前节点
fwrite($this->indexHandler, pack('L', $indexOffset), 4);
fseek($this->indexHandler, 0, SEEK_END);
//写入当前索引节点
fwrite($this->indexHandler, $dataBlock, DB_INDEX_SIZE);
fseek($this->dataHandler, 0, SEEK_END);
//将新数据写入数据文件
fwrite($this->dataHandler, $data, strlen($data));
return DB_INSERT_SUCCESS;
}

$foundFlag = false;

while ($position) {
fseek($this->indexHandler, $position, SEEK_SET);
//获取当前索引节点的值
$tmpBlock = fread($this->indexHandler, DB_INDEX_SIZE);
$currentKey = substr($tmpBlock, 4, DB_KEY_SIZE);
//因为索引文件是二进制值，使用strncmp函数比较是否相等。
if (!strncmp($key, $currentKey, strlen($key))) {
//当前索引指向的数据在数据文件中的偏移量
$dataOff = unpack('L', substr($tmpBlock, DB_KEY_SIZE + 4, 4));
$dataOff = $dataOff[1];
//当前索引指向的数据长度
$dataLe = unpack('L', substr($tmpBlock, DB_KEY_SIZE + 8, 4));
$dataLe = $dataLe[1];
$foundFlag = true;
break;
}
$prev = $position;
$position = unpack('L', substr($tmpBlock, 0, 4));
$position = $position[1];
}

if ($foundFlag) {
return DB_EXISTS_KEY;
}

fseek($this->indexHandler, $prev, SEEK_SET);
//上一个节点指向当前节点
fwrite($this->indexHandler, pack('L', $indexOffset), 4);
fseek($this->indexHandler, 0, SEEK_END);
////写入当前索引节点
fwrite($this->indexHandler, $dataBlock, DB_INDEX_SIZE);
fseek($this->dataHandler, 0, SEEK_END);
//将新数据写入数据文件
fwrite($this->dataHandler, $data, strlen($data));
return DB_INSERT_SUCCESS;
}

public function find($key)
{
$offset = $this->hashFunc($key) % DB_BUCKET_SIZE * 4;
fseek($this->indexHandler, $offset, SEEK_SET);
$position = unpack('L', fread($this->indexHandler, 4));
$position = $position[1];

$foundFlag = false;
while ($position) {
fseek($this->indexHandler, $position, SEEK_SET);
$indexBlock = fread($this->indexHandler, DB_INDEX_SIZE);
$currentKey = substr($indexBlock, 4, DB_KEY_SIZE);
if (!strncmp($currentKey, $key, strlen($key))) {
$dataOffset = unpack('L', substr($indexBlock, DB_KEY_SIZE + 4, 4));
$dataOffset = $dataOffset[1];

$dataLen = unpack('L', substr($indexBlock, DB_KEY_SIZE + 8, 4));
$dataLen = $dataLen[1];

$foundFlag = true;
break;
}

$position = unpack('L', substr($indexBlock, 0, 4));
$position = $position[1];
}

if ($foundFlag) {
fseek($this->dataHandler, $dataOffset, SEEK_SET);
$data = fread($this->dataHandler, $dataLen);
return $data;
} else {
return NULL;
}

}

public function delete($key)
{
$offset = $this->hashFunc($key) % DB_BUCKET_SIZE * 4;
fseek($this->indexHandler, $offset, SEEK_SET);
$head = unpack('L', fread($this->indexHandler, 4));
$head = $head[1];

$current = $head;
$prev = 0;

$foundFlag = false;

while ($current) {
fseek($this->indexHandler, $current, SEEK_SET);
$dataBlock = fread($this->indexHandler, DB_INDEX_SIZE);

$currentKey = substr($dataBlock, 4, DB_KEY_SIZE);

$next = unpack('L', substr($dataBlock, 0, 4));
$next = $next[1];

if (!strncmp($key, $currentKey, strlen($key))) {
$foundFlag = true;
break;
}

$prev = $current;
$current = $next;
}

if (!$foundFlag) {
return DB_DELETET_FAILED;
}

if ($prev == 0) {
fseek($this->indexHandler, $offset, SEEK_SET);
} else {
fseek($this->indexHandler, $prev, SEEK_SET);
}

//把上一个索引指向下一个索引的指针
//指向要删除的索引指向的下一个索引，
//从而完成删除操作，并未直接删除索引和数据文件中的值
fwrite($this->indexHandler, pack('L', $next), 4);

return DB_DELETE_SUCCESS;
}

protected function hashFunc($str)
{
$str = substr(md5($str), 0, 8);
$hashValue = 0;

for ($i = 0; $i < 8; $i++) {
$hashValue += 33 * $hashValue + ord($str[$i]);
}

return $hashValue & 0x7FFFFFFF;
}
}

github源码地址