如何用Go实现事务型的KV存储

如果您想设计一个交互式shell，允许访问内存中的事务型键/值存储，那么这篇文章就是你想要的。

我们一起来设计一个事务型的KV存储。

背景

设计这个事务型KV存储的灵感来源于一位网友分享了他30天马拉松式面试的经验，非常精彩。我了解到他在面试中被问到的这个有趣的系统设计问题。

挑战

该系统设计面临的问题有：
构建一个交互式shell，允许访问“内存中的事务型键/值存储”。
shell能够接收以下命令：

存在问题

Q1: 交互shell会话结束后，数据是否持久化？
Q2: 对数据的操作是否反映到全局shell?
Q3: 在嵌套事务中提交更改是否也反映到上一级级事务中?
你的问题可能不同。你问的问题越多，你就越能理解问题。
这个设计的处理很大程度上取决于所问的问题，所以在我们完成这个设计之前定义几个前提条件：

1. 数据不持久化（也就是，shell会话关闭，数据丢失）
2. KV的值只有string类型（可以使用接口来自定义类型，但不在本文范围内）
   下面让我们试着理解问题中棘手的部分。

理解"事务"

使用BEGIN命令创建一个事务，并为将要发生的其他操作创建上下文。例如:

> BEGIN // 创建一个事务
> SET X 200
> SET Y 14
> GET Y
14

这是当前活动事务，所有操作只在其中工作。只有活动事务通过COMMIT命令被提交，这些操作才生效。ROLLBACK命令会将这个活动事务产生的操作都丢弃。准确的说，从map中删除所有的键值对。
例如：

> BEGIN //创建当前活动事务
> SET Y 2020
> GET Y
2020
> ROLLBACK //丢弃前面的操作
> GET Y
Y not set // 设置Y的值被丢弃

事务可能有嵌套，也就是包含子事务：

事务嵌套

新生成的事务从其父事务继承变量，在子事务上下文中所做的更改也将反映在父事务中。
例如：

> BEGIN //创建一个新的活动事务
> SET X 5
> SET Y 19
> BEGIN //在前面的事务上下文中创建新的事务，也是当前活动事务
> GET Y
Y = 19 //新事务继承其父事务的上下文**
> SET Y 23
> COMMIT //Y的新值已被持久化到键值存储区**
> GET Y
Y = 23 // SET Y=19所做的更改已被丢弃**

开始设计

我们讨论的事务还包含子事务，可以使用栈数据结构来存储：

事务栈可视化

• 栈的每个元素是事务。
• 栈顶部存储当前活动事务。
• 每个事务包含自己的键值对Map。我们称为“本地存储”类似本地缓存，当SET一个变量到当前事务，这个存储就更新。
• 一旦事务中的修改提交，本地存储的值就写入全局map对象中。
  我们将使用链表来实现栈。也可以使用动态数组来实现，这个留给读者作为作业。

package designPattern

//GlobalStore存储全局KV
var GlobalStore = make(map[string]string)

//Transaction
type Transaction struct {
    store map[string]string //每个事务有自己的本地存储
    next *Transaction
}

//TransactionStack存储事务元素
type TransactionStack struct {
    top *Transaction
    size int
}

• 栈用结构体来表示，top表示栈顶元素；size表示栈包含的事务个数，该字段可选。
• Transaction结构体包含存储事务map和指向下一个事务指针。
• GlobalStore是一个map，所有的事务都可以使用这个map来存储提交后的KV。
  下面实现TransactionStack栈的pop和push方法。

//PushTransaction创建一个活动事务
func (ts *TransactionStack) PushTransaction() {
    temp := Transaction{store: make(map[string]string)}
    temp.next = ts.top
    ts.top = &temp
    ts.size++
}

//PopTransaction删除栈的一个事务
func (ts *TransactionStack)PopTransaction()  {
    if ts.top == nil {
        fmt.Printf("ERROR: No Active Transaction\n")
    } else{
        ts.top = ts.top.next
        ts.size--
    }
}

• 每个BEGIN操作，就向TransactionStack栈push一个事务，更新栈顶值。
• 遇到COMMIT或END操作，就将当前活动事务从栈弹出，因此活跃事务就转成前一个事务（父事务）。
  下面在创建一个Peek方法返回栈顶部事务：

//Peek 返回活跃事务
func (ts *TransactionStack)PeeK() *Transaction {
    return ts.top
}

注意我们返回一个指向事务的指针。
执行COMMIT命令会将当前活动事务的新增或更新的KV值复制到全局变量GlobalStore中。

//Commit 将TransactionStack顶部事务包含新增/更新KV存储到文件或磁盘中进行持久化
func (ts *TransactionStack) Commit()  {
    ActiveTransaction := ts.PeeK()
    if ActiveTransaction != nil {
        for key, value := range ActiveTransaction.store{
            GlobalStore[key] = value
            if ActiveTransaction.next != nil {
                //子事务可能更新父事务的内容，所以需要将子事务KV复制给父事务
                ActiveTransaction.next.store[key] = value
            }
        }
    } else {
        fmt.Printf("INFO: Nothing to commit\n")
    }
    //如果需要持久化，将数据写入文件，提示：将map序列化为json
}

回退一个事务非常简单。只需要将对应事务的本地存储Map中的KV删除即可。

//RollBackTransaction 清除该事务中的所有Keys
func (ts *TransactionStack)RollBackTransaction()  {
    if ts.top == nil {
        fmt.Printf("ERROR: No Active Transaction\n")
    } else {
        for key := range ts.top.store {
            delete(ts.top.store, key)
        }
    }
}

最后，事务中的SET和GET操作：

//GET 从本地存储Map中读取key
func Get(key string, T *TransactionStack)  {
    ActiveTransaction := T.PeeK()
    if ActiveTransaction == nil {
        if val, ok :=  GlobalStore[key]; ok {
            fmt.Printf("%s\n", val)
        }else{
            fmt.Printf("%s not set\n", key)
        }
    } else {
        if val, ok := ActiveTransaction.store[key]; ok {
            fmt.Printf("%s\n", val)
        } else {
            fmt.Printf("%s not set\n", key)
        }
    }
}

然而SET操作，我们还需要考虑用户可能没有执行事务的情况。这意味着我们的栈是空的，也就是用户直接在全局变量中设置KV值。

> SET F 55
> GET F
55

这种情况我们可以直接更新GlobalStore：

//Set key
func Set(key string, value string, t *TransactionStack)  {
    //从活跃事务中读取key是否存在
    ActiveTransaction := t.PeeK()
    if ActiveTransaction == nil {
        GlobalStore[key] = value
    } else {
        ActiveTransaction.store[key] = value
    }
}

我们已经基本完成了键值存储功能，接下来让我们写主程序代码:

func main() {
    reader := bufio.NewReader(os.Stdin)
    items := &TransactionStack{}
    for {
        fmt.Printf(">")
        text, _ := reader.ReadString('\n')
        //将文本拆分为操作字符串
        operation := strings.Fields(text)
        switch operation[0] {
        case "BEGIN":
            items.PushTransaction()
        case "ROLLBACK":
            items.RollBackTransaction()
        case "COMMIT":
            items.Commit()
            items.PopTransaction()
        case "END":
            items.PopTransaction()
        case "SET":
            Set(operation[1], operation[2], items)
        case "GET":
            Get(operation[1], items)
        case "DELETE":
            Delete(operation[1], items)
        case "COUNT":
            Count(operation[1], items)
        case "STOP":
            os.Exit(0)
        default:
            fmt.Printf("ERROR: Unrecognised Operation %s\n", operation[0])
        }
    }
}

如果你理解了前面的代码，COUNT和DELET的实现是非常简单的读者可以自己动手实现。
下面看下运行效果：

>SET R 11
>GET R  
11
>DELETE R
>GET R
R not set
>BEGIN
>SET F 82
>SET R 90
>GET R
90
>COMMIT
>GET R
90
>END
ERROR: No Active Transaction
>COUNT 90
1
>Count 88
0
>STOP

Process finished with exit code 0