從零動手寫數據庫系統：數據庫系統的日志模塊實現
任何一個應用只要冠以”系統“二字，那么它一定離不開一個模塊，那就是”日志“。既然我們要開發一個數據庫系統，那么它必然要有自己的日志模塊。日志通常用于記錄系統的運行狀態，有點類似于快照，一旦系統出現異常，那么管理員或者它的代碼本身可以通過掃描分析日志來確定問題所在，或者通過日志執行錯誤恢復，這點對數據庫系統更加重要。
數據庫系統經常要往文件中讀寫大量數據，在這個過程中很容易出現各種各樣的問題，例如在執行一個交易時，網絡突然斷開，機器突然斷電，于是交易執行到一半就會突然中斷，當系統重新啟動時，整個數據庫就會處于一種錯誤狀態，也就是有一部數據寫入，但還有一部分數據丟失，這種情況對數據庫系統而言非常致命，倘若不能保證數據的一致性，那么這種數據系統就不會有人敢使用。那如何保證數據一致性呢，這就得靠日志來保證，數據庫在讀寫數據前，會先寫入日志，記錄相應的操作，例如當前操作是讀還是寫，然后記錄要讀寫的數據。
假設我們現在有個業務，要把一百行數據寫入兩個表，前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_manager
import (
fm "file_manager"
"sync"
)
const (
UINT64_LEN = 8
)
type LogManager struct {
file_manager *fm.FileManager
log_file string
log_page *fm.Page
current_blk *fm.BlockId
latest_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_LEN
var err error
if 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_manager
import (
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.boundary
return 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_manager
import (
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 一個是數值i
rec := 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”，這個字符串會以字節數組的方式寫入到區塊中，然后再把讀入的數據重新讀取，同時判斷讀取的數據與寫入的是否一致。
