自己动手写数据库：数据库系统的日志模块实现

任何一个应用只要冠以”系统“二字，那么它一定离不开一个模块，那就是”日志“。既然我们要开发一个数据库系统，那么它必然要有自己的日志模块。日志通常用于记录系统的运行状态，有点类似于快照，一旦系统出现异常，那么管理员或者它的代码本身可以通过扫描分析日志来确定问题所在，或者通过日志执行错误恢复，这点对数据库系统更加重要。

数据库系统经常要往文件中读写大量数据，在这个过程中很容易出现各种各样的问题，例如在执行一个交易时，网络突然断开，机器突然断电，于是交易执行到一半就会突然中断，当系统重新启动时，整个数据库就会处于一种错误状态，也就是有一部数据写入，但还有一部分数据丢失，这种情况对数据库系统而言非常致命，倘若不能保证数据的一致性，那么这种数据系统就不会有人敢使用。那如何保证数据一致性呢，这就得靠日志来保证，数据库在读写数据前，会先写入日志，记录相应的操作，例如当前操作是读还是写，然后记录要读写的数据。假设我们现在有个业务，要把一百行数据写入两个表，前50行写入表1，后50行写入表2，于是日志就会记录”表1写入0到50行“；”表2写入51到100行“，这类信息。假设在数据写入前50行后突然断电，机器重启，数据库系统重新启动后，它自动扫描日志发现”表2写入51到100行“这个操作没有执行，于是再次执行这个操作，这样数据的一致性就能得以保证。

本节我们在上一节实现文件系统的基础上，看看如何实现日志模块。对于日志模块而言，日志就是一组字节数组，它只负责把数组内容写入内存或是磁盘文件，数据到底有什么内容，格式如何解析它一概不管。同时在日志写入时采用”压栈“模式，假设我们有3条日志，其长度分别为50字节，100字节，100字节，现在我们有400字节的缓存可以写入，那么写入日志时我们将从缓存的末尾开始写，例如存入第一条日志时，我们从缓存第350字节开始写入，于是350字节到400字节就对应第一条日志，然后我们把当前可写入的地址放置到缓存的开头8字节，例如第一条日志写入后，下次可写入的地址是350，于是我们在缓存开头8字节存入数据350，当要写入第二条日志时，我们读取缓存前8字节，拿到数值350，由于第二条缓存长度100字节，于是我们将其写入缓存的地址为350-100=250，于是写入后缓存的250到350部分的内容就对应第二条日志，然后我们将250写入缓存开头8字节；当写入第三条日志时，系统读取开头8字节得到数值250，于是第三条日志的写入地址就是250-100=150，于是系统将第三条日志写入缓存偏移150字节处，于是从150字节到250字节这部分的数据就对应第3条日志，同时把150存入开头8字节，以此类推。

废话不多说，我们看看具体代码实现，首先创建文件夹log_manager，加入log_manager.go，输入以下代码：

package log_managerimport (fm "file_manager""sync"
)const (UINT64_LEN = 8
)type LogManager struct {file_manager   *fm.FileManagerlog_file       stringlog_page       *fm.Pagecurrent_blk    *fm.BlockIdlatest_lsn     uint64 //当前日志序列号last_saved_lsn uint64 //上次存储到磁盘的日志序列号mu             sync.Mutex
}func (l *LogManager) appendNewBlock() (*fm.BlockId, error) {blk, err := l.file_manager.Append(l.log_file)if err != nil {return nil, err}/*添加日志时从内存的底部往上走，例如内存400字节，日志100字节，那么日志将存储在内存的300到400字节处，因此我们需要把当前内存可用底部偏移写入头8个字节*/l.log_page.SetInt(0, uint64(l.file_manager.BlockSize()))l.file_manager.Write(&blk, l.log_page)return &blk, nil
}func NewLogManager(file_manager *fm.FileManager, log_file string) (*LogManager, error) {log_mgr := LogManager{file_manager:   file_manager,log_file:       log_file,log_page:       fm.NewPageBySize(file_manager.BlockSize()),last_saved_lsn: 0,latest_lsn:     0,}log_size, err := file_manager.Size(log_file)if err != nil {return nil, err}if log_size == 0 { //如果文件为空则添加新区块blk, err := log_mgr.appendNewBlock()if err != nil {return nil, err}log_mgr.current_blk = blk} else { //文件有数据，则在文件末尾的区块读入内存，最新的日志总会存储在文件末尾log_mgr.current_blk = fm.NewBlockId(log_mgr.log_file, log_size-1)file_manager.Read(log_mgr.current_blk, log_mgr.log_page)}return &log_mgr, nil
}func (l *LogManager) FlushByLSN(lsn uint64) error {/*将给定编号及其之前的日志写入磁盘，注意这里会把与给定日志在同一个区块，也就是Page中的日志也写入磁盘。例如调用FlushLSN(65)表示把编号65及其之前的日志写入磁盘，如果编号为66,67的日志也跟65在同一个Page里，那么它们也会被写入磁盘*/if lsn > l.last_saved_lsn {err := l.Flush()if err != nil {return err}l.last_saved_lsn = lsn}return nil
}func (l *LogManager) Flush() error {//将当前区块数据写入写入磁盘_, err := l.file_manager.Write(l.current_blk, l.log_page)if err != nil {return err}return nil
}func (l *LogManager) Append(log_record []byte) (uint64, error) {//添加日志l.mu.Lock()defer l.mu.Unlock()boundary := l.log_page.GetInt(0) //获得可写入的底部偏移record_size := uint64(len(log_record))bytes_need := record_size + UINT64_LENvar err errorif int(boundary-bytes_need) < int(UINT64_LEN) {//当前容量不够,先将当前日志写入磁盘err = l.Flush()if err != nil {return l.latest_lsn, err}//生成新区块用于写新数据l.current_blk, err = l.appendNewBlock()if err != nil {return l.latest_lsn, err}boundary = l.log_page.GetInt(0)}record_pos := boundary - bytes_need         //我们从底部往上写入l.log_page.SetBytes(record_pos, log_record) //设置下次可以写入的位置l.log_page.SetInt(0, record_pos)l.latest_lsn += 1 //记录新加入日志的编号return l.latest_lsn, err
}func (l *LogManager) Iterator() *LogIterator {//生成日志遍历器l.Flush()return NewLogIterator(l.file_manager, l.current_blk)
}

上面代码所构造的日志管理器，其作用就是将写入的日志先存储在内存块中，一旦当前内存块写满则将其写入磁盘文件，然后生成新的内存块用于写入新日志。每次日志管理器启动时，它根据给定的目录读取目录下的二进制文件，将文件尾部的区块读入内存，这样就能得到文件存储的日志数据。

问了更好的遍历日志，我们要构造一个日志遍历器，在同一个目录下创建log_iterator.go，然后写入以下内容：

package log_managerimport (fm "file_manager"
)/*
LogIterator用于遍历给定区块内的记录,由于记录从底部往上写，因此记录1,2,3,4写入后在区块的排列为
4,3,2,1，因此LogIterator会从上往下遍历记录，于是得到的记录就是4,3,2,1
*/type LogIterator struct {file_manager *fm.FileManager blk         *fm.BlockId p           *fm.Page current_pos uint64 boundary    uint64
}func NewLogIterator(file_manager *fm.FileManager, blk *fm.BlockId) *LogIterator{it := LogIterator{file_manager: file_manager,blk: blk , }//现将给定区块的数据读入it.p = fm.NewPageBySize(file_manager.BlockSize())err := it.moveToBlock(blk) if err != nil {return nil }return &it
}func (l *LogIterator) moveToBlock(blk *fm.BlockId) error {//打开存储日志数据的文件，遍历到给定区块，将数据读入内存_, err := l.file_manager.Read(blk, l.p)if err != nil {return err }//获得日志的起始地址l.boundary = l.p.GetInt(0)l.current_pos = l.boundaryreturn nil
}func (l *LogIterator) Next() []byte {//先读取最新日志，也就是编号大的，然后依次读取编号小的if l.current_pos == l.file_manager.BlockSize() {l.blk = fm.NewBlockId(l.blk.FileName(), l.blk.Number() - 1)l.moveToBlock(l.blk)}record := l.p.GetBytes(l.current_pos)l.current_pos += UINT64_LEN + uint64(len(record))return record
}func (l *LogIterator) HasNext() bool {//如果当前偏移位置小于区块大那么还有数据可以从当前区块读取//如果当前区块数据已经全部读完，但是区块号不为0，那么可以读取前面区块获得老的日志数据return l.current_pos < l.file_manager.BlockSize() || l.blk.Number() > 0
}

日志遍历器的作用是逐条读取日志，它先从最新的日志开始读取，然后依次获取老的日志。最后我们通过测试用例来理解当前代码的作用和逻辑，添加log_manager_test.go，代码如下：

package log_managerimport (fm "file_manager""fmt""github.com/stretchr/testify/require""testing"
)func makeRecord(s string, n uint64) []byte {//使用page提供接口来设置字节数组的内容p := fm.NewPageBySize(1)npos := p.MaxLengthForString(s)b := make([]byte, npos+UINT64_LEN)p = fm.NewPageByBytes(b)p.SetString(0, s)p.SetInt(npos, n)return b
}func createRecords(lm *LogManager, start uint64, end uint64) {for i := start; i <=end; i++ {//一条记录包含两个信息，一个是字符串record 一个是数值irec := makeRecord(fmt.Sprintf("record%d", i), i)lm.Append(rec)}
}func TestLogManager(t *testing.T) {file_manager, _ := fm.NewFileManager("logtest", 400)log_manager, err := NewLogManager(file_manager, "logfile")require.Nil(t, err)createRecords(log_manager, 1, 35)iter := log_manager.Iterator()rec_num := uint64(35)for iter.HasNext() {rec := iter.Next()p := fm.NewPageByBytes(rec)s := p.GetString(0)require.Equal(t, fmt.Sprintf("record%d", rec_num), s)npos := p.MaxLengthForString(s)val := p.GetInt(npos)require.Equal(t, val, rec_num)rec_num -= 1}createRecords(log_manager, 36, 70)log_manager.FlushByLSN(65)iter = log_manager.Iterator()rec_num = uint64(70)for iter.HasNext() {rec := iter.Next()p := fm.NewPageByBytes(rec)s := p.GetString(0)require.Equal(t, fmt.Sprintf("record%d", rec_num), s)npos := p.MaxLengthForString(s)val := p.GetInt(npos)require.Equal(t, val, rec_num)rec_num -= 1}
}

用例首先创建日志管理器，然后写入35条日志，每条日志后面又跟着日志的编号。例如第35条日志的内容为"record35"，这个字符串会以字节数组的方式写入到区块中，然后再把读入的数据重新读取，同时判断读取的数据与写入的是否一致。

更详细的讲解和调试演示，请在B站搜索Coding迪斯尼，更多有趣内容请点击这里：http://m.study.163.com/provider/7600199/index.htm?share=2&shareId=7600199,代码下载链接：https://github.com/wycl16514/data-system-log-manager.git

自己动手写数据库：数据库系统的日志模块实现相关推荐

自己动手写一个nodejs的日志生成器
自己动手写一个nodejs的logger 最近正在边学边用node.js开发个人应用的server,由于有用到websocket相关,想对websocket的通信选择性的做下日志记录,所以萌发了自己动 ...
自己动手写数据库:并发管理的基本原理
一个好的数据库,其特点必然是吞吐量高,也就是它能在高并发请求压力下保证数据的准确性和安全性,由此并发管理是不可或缺的一环.事实上并发管理是一个相当复杂的计算机科学领域的课题,它几乎可以自成一格领域,是 ...
自己动手写数据库：记录管理器的设计和实现
在数据库中,数据以"记录"作为一个单元来存储,例如一个表的"一行"就对应一条记录.假设我们有一个表叫STUDENT,其中有name, age, sex, cla ...
python写日志文件_Python logging日志模块配置文件方式
在一些微服务或web服务中我们难免需要日志功能,用来记录一些用户的登录记录,操作记录,以及一些程序的崩溃定位,执行访问定位等等; Python内置非常强大的日志模块 ==> logging 今 ...
Linux C语言自己动手写日志生成函数
有时候需要自己把日志信息保存到日志文件中,没有找到现成的函数,只好自己动手写一个,完成相关功能. void LOG(const char* ms, ... ) {char wzLog[1024] = ...
服务器的日志用数据库还是文件,日志是写数据库好还是文件好，以及日志策略...
日志是写数据库好还是文件好,以及日志策略电脑版发表于:2019/6/4 21:36 先来看看大佬的一句话:离开需求谈技术都是扯淡主要是看你的日志是为了只做一个存储归档,还是要把数据对统计与分析如 ...
数据库运维平台~慢日志模块设计
一目的:统一对慢日志进行收集并且展示二工具: percona工具组件三设计: mysql数据库部署慢日志脚本实现两个目的 1 慢日志进行切割 2自动插入到远程数据库实现统一收集处理四 ...
php怎么自己写框架,PHP学习笔记,自己动手写个MVC的框架
最新在大家自己的博客的过程中,发现各种开源的博客系统都或多或少的用起来别扭.于是想动手自己写个博客系统.既然写,就想好好写.那就先写个MVC框架.一点一点来.写的过程中有很多想法.还希望大家能够多多指 ...
自己动手写一个推荐系统,推荐系统小结,推荐系统：总体介绍、推荐算法、性能比较, 漫谈“推荐系统”, 浅谈矩阵分解在推荐系统中的应用...
自己动手写一个推荐系统废话: 最近朋友在学习推荐系统相关,说是实现完整的推荐系统,于是我们三不之一会有一些讨论和推导,想想索性整理出来. 在文中主要以工程中做推荐系统的流程着手,穿插一些经验之谈,并 ...

自己动手写数据库：数据库系统的日志模块实现

自己动手写数据库：数据库系统的日志模块实现相关推荐

最新文章

热门文章