谈一谈如何使用etcd中的事务以及自己的理解

手机

iphone11大小尺寸是多少？苹果iPhone11和iPhone13的区别是什么？

警方通报辅警执法直播中被撞飞：犯罪嫌疑人已投案

家电

谈一谈如何使用etcd中的事务以及自己的理解

2023-05-08 22:15:22 来源：博客园

01、谈一谈etcd事务的如何使用以及自己的理解

本文内容来源于自己学习时所做的记录，主要来源于文章最后的参考链接，如有侵权，请联系删除，谢谢！

【资料图】

etcd 是一个 key/value 类型的数据库。既然我们需要存储数据，必然会面临这样一个需求，即希望无论什么样的场景下，一组操作要么同时完成，要么都失败，哪怕数据库出现了故障，甚至了机器发生了宕机。幸运的是从 etcd 3.4 版本开始，开始支持事务。

etcd 中的事务，实现了原子地执行冲突检查、更新多个 keys 的值。除此之外，etcd 将底层 MVCC 机制的版本信息暴露出来，基于版本信息封装出了一套基于乐观锁的事务框架 STM，并实现了不同的隔离级别。接下来我们就一起学习学习。

1、什么是事务？

事务（Transaction）是指数据库管理系统中的一个基本操作单位，通常由一组对数据库的读/写操作组成一个不可分割的操作序列。

在关系型数据库中，事务应该具备 ACID特性，即原子性、一致性、隔离性、持久性。具体解释如下：

原子性（Atomicity）：事务作为一个不可分割的操作序列，要么全部执行成功，要么全部回滚。如果事务在执行过程中发生错误，需要将已经修改的数据撤销到事务开始前的状态。
一致性（Consistency）：指事务执行前和执行后的数据状态保持一致。在事务执行过程中，数据库会处于一种中间状态，需要通过在事务结束时进行一些额外的工作来确保数据的一致性。这里理解起来还是有点抽象，我们借助一个例子来说明:

比如，用户 A 和用户 B 在银行分别有 1000 元和 500 元，总共 1500 元，用户 A 给用户 B 转账 200 元，分为两个步骤，从 A 的账户扣除 200 元和对 B 的账户增加 200 元。一致性就是要求上述步骤操作后，最后的结果是用户 A 还有 800 元，用户 B 有 700 元，总共 1500 元，而不会出现用户 A 扣除了 200 元，但用户 B 未增加的情况（该情况，用户 A 为 800 元，用户 B 为500 元，总共 1300 元）。

隔离性（Isolation）：多个事务并发执行时，每个事务是彼此隔离的，互相不干扰。隔离级别越高，就越能防止出现脏读、不可重复读、幻读等问题。
持久性（Durability）：指一旦事务提交，则所有修改的数据都将永久保存在数据库中。即使发生系统故障，也不会丢失任何数据。

常见的关系型数据库如 MySQL，在存储引擎为 InnoDB时是支持事务的，那InnoDB 引擎通过什么技术来保证事务的这四个特性的呢？

持久性是通过 redo log （重做日志）来保证的；
原子性是通过 undo log（回滚日志）来保证的；
隔离性是通过 MVCC（多版本并发控制）或锁机制来保证的；
一致性则是通过持久性+原子性+隔离性来保证；

了解了 MySQL 中的事务的简单实现方式后，接下来讲一讲 etcd 中的事务。

etcd 中的事务则是基于 CAS（Compare and Swap，即比较再交换）方式。

etcd 使用了不到四百行的代码实现了迷你事务，其对应的语法为 If-Then-Else。etcd 允许用户在一次修改中批量执行多个操作，即这一组操作被绑定成一个原子操作，并共享同一个修订号。写法类似 CAS，如下所示：

Txn().If(cond1, cond2, ...).Then(op1, op2, ...,).Else(op1, op2)

上面语句的意思是，如果 If 冲突判断语句为真，对应返回值为 true，Then 中的语句将会被执行，否则执行 else 中的逻辑。

在 etcd 事务执行过程中，客户端与 etcd 服务端之间没有维护事务会话。冲突判断(If)和执行过程 Then/Else作为一个原子过程来执行 If-Then-Else，因此 etcd 事务不会发生阻塞，无论成功还是失败都会返回，当发生冲突导致执行失败时，需要应用进行重试。业务代码需要考虑这部分的重试逻辑。

讲了这么多理论的东西，接下来就一起看一看 etcd 中事务的使用案例。

2、etcd中事务的使用案例

这里使用一个比较容易理解的例子来进行演示，用户A 向用户B 转账的场景。etcd 的事务基于乐观锁思想，当检测到冲突会发起重试，检测冲突时使用了ModRevision进行校验，该字段表示某个 key 上一次被更改时，全局的版本是多少。因此，我们实现转账业务的流程如下所示：

代码如下：

package mainimport ("context""fmt"clientv3 "go.etcd.io/etcd/client/v3""strconv""time")func main() {config := clientv3.Config{Endpoints:   []string{"192.168.91.66:12379"},DialTimeout: 5 * time.Second,}// 建立连接client, err := clientv3.New(config)if err != nil {fmt.Println(err)return}sender, receiver := "/sender_amount", "/receiver_amount"_, err = client.Put(context.Background(), sender, "1000")if err != nil {fmt.Printf("etcd put /sender_amount failed, err:%v\n", err)return}_, err = client.Put(context.Background(), receiver, "500")if err != nil {fmt.Printf("etcd put /receiver_amount failed, err:%v\n", err)return}err = txnTransfer(client, sender, receiver, 200)if err != nil {fmt.Printf("etcd txnTransfer failed, err:%v\n", err)return}}func txnTransfer(etcd *clientv3.Client, sender, receiver string, amount uint64) error {// 失败重试for {if ok, err := doTxn(etcd, sender, receiver, amount); err != nil {return err} else if ok {return nil}}}func doTxn(etcd *clientv3.Client, sender, receiver string, amount uint64) (bool, error) {// 第一个事务，利用事务的原子性，同时获取发送和接收者的余额以及 ModRevisiongetResp, err := etcd.Txn(context.TODO()).Then(clientv3.OpGet(sender), clientv3.OpGet(receiver)).Commit()if err != nil {return false, err}senderKV := getResp.Responses[0].GetResponseRange().Kvs[0]receiverKV := getResp.Responses[1].GetResponseRange().Kvs[0]senderNum, receiverNum := toUInt64(senderKV.Value), toUInt64(receiverKV.Value)// 验证账户余额是否充足if senderNum < amount {return false, fmt.Errorf("资金不足")}// 发起转账事务，冲突判断 ModRevision 是否发生变化txn := etcd.Txn(context.TODO()).If(clientv3.Compare(clientv3.ModRevision(sender), "=", senderKV.ModRevision),clientv3.Compare(clientv3.ModRevision(receiver), "=", receiverKV.ModRevision))// ModRevision 未发生变化，即 If 判断条件成功txn = txn.Then(clientv3.OpPut(sender, fromUint64(senderNum-amount)),     // 更新发送者账户余额clientv3.OpPut(receiver, fromUint64(receiverNum+amount))) // 更新接收者账户余额resp, err := txn.Commit() // 提交事务if err != nil {return false, err}return resp.Succeeded, nil}func toUInt64(value []byte) uint64 {ret, err := strconv.Atoi(string(value))if err != nil {fmt.Printf("value to uint64 failed, err: %v\n", err)return 0}return uint64(ret)}func fromUint64(number uint64) string {return strconv.Itoa(int(number))}

转账业务开始前，sender、reciver的余额如下：

转账业务完成后，sender、reciver的余额如下：

如上 etcd 事务的实现基于乐观锁思想，涉及到两次事务操作，第一次事务利用原子性来同时获取发送方和接收方的当前账户金额；第二次事务发起转账操作，冲突检测 ModRevision是否发生变化，如果没有变化则正常提交事务。若发生了冲突，则需要进行重试。

如上过程的实现较为繁琐，除了业务逻辑，还有大量的代码用来判断冲突以及重试。因此，etcd 社区基于事务特性，实现了一个简单的事务框架 STM，构建了各个事务隔离级别类，下面我们看看基于 STM 框架如何实现 etcd 事务。

3、使用 STM 实现转账

那如何简化 etcd 事务实现的过程呢，etcd clientv3 提供了 STM，即软件事务内存，帮我们自动处理这些繁琐的过程。使用 STM 的转账业务代码如下：

package mainimport ("context""fmt"clientv3 "go.etcd.io/etcd/client/v3""go.etcd.io/etcd/client/v3/concurrency""strconv""time")func main() {config := clientv3.Config{Endpoints:   []string{"192.168.91.66:12379"},DialTimeout: 5 * time.Second,}// 建立连接client, err := clientv3.New(config)if err != nil {fmt.Println(err)return}sender, receiver := "/sender_amount", "/receiver_amount"_, err = client.Put(context.Background(), sender, "1000")if err != nil {fmt.Printf("etcd put /sender_amount failed, err:%v\n", err)return}_, err = client.Put(context.Background(), receiver, "500")if err != nil {fmt.Printf("etcd put /receiver_amount failed, err:%v\n", err)return}err = txnStmTransfer(client, sender, receiver, 200)if err != nil {fmt.Printf("etcd txnTransfer failed, err:%v\n", err)return}}func txnStmTransfer(cli *clientv3.Client, from, to string, amount uint64) error {// NewSTM 创建了一个原子事务的上下文，并把我们的业务代码作为一个函数传进去_, err := concurrency.NewSTM(cli, func(stm concurrency.STM) error {// stm.Get 封装好了事务的读操作senderNum := toUint64(stm.Get(from))receiverNum := toUint64(stm.Get(to))if senderNum < amount {return fmt.Errorf("余额不足")}// stm.Put封装好了事务的写操作stm.Put(to, fromUint64(receiverNum+amount))stm.Put(from, fromUint64(senderNum-amount))return nil})return err}func toUint64(value string) uint64 {ret, err := strconv.Atoi(value)if err != nil {fmt.Printf("value to toUint64 failed, err: %v\n", err)return 0}return uint64(ret)}func fromUint64(number uint64) string {return strconv.Itoa(int(number))}

上述基于 STM 实现的转账业务流程，我们只要关注转账逻辑的实现即可，事务相关的其他操作由 STM 完成。

STM 的使用特别简单，只需把业务相关的代码封装成可重入的函数传给 stm，然后 STM 会处理好其余所有的细节。STM 对象在内部构造 txn 事务，把我们编写的业务函数翻译成 If-Then，自动提交事务，处理失败重试等工作，直到事务执行成功，或者出现异常，重试亦不能解决。

接下来我们一起初略的看看 NewSTM的内部做了些什么

// 位于 clientv3/concurrency/stm.go:89func NewSTM(c *v3.Client, apply func(STM) error, so ...stmOption) (*v3.TxnResponse, error) {   opts := &stmOptions{ctx: c.Ctx()}   for _, f := range so {      f(opts)   }   if len(opts.prefetch) != 0 {      f := apply      apply = func(s STM) error {         s.Get(opts.prefetch...)         return f(s)      }   }   return runSTM(mkSTM(c, opts), apply)}

根据源码可以知道，NewSTM首先创建一个 stm，然后执行 stm，代码如下所示：

func runSTM(s STM, apply func(STM) error) (*v3.TxnResponse, error) {outc := make(chan stmResponse, 1)go func() {defer func() {if r := recover(); r != nil {e, ok := r.(stmError)if !ok {// client apply panickedpanic(r)}outc <- stmResponse{nil, e.err}}}()var out stmResponsefor {s.reset()if out.err = apply(s); out.err != nil {break}if out.resp = s.commit(); out.resp != nil {break}}outc <- out}()r := <-outcreturn r.resp, r.err}

runstm 主要是循环执行以下三个步骤：

重置 stm，清空 STM 的读写缓存
执行事务操作，apply 函数
提交事务

etcd client 最终执行提交事务的操作：

txnresp, err := s.client.Txn(s.ctx).If(s.conflicts()...).Then(s.wset.puts()...).Commit()

4、STM 实现细节

下面我们来看 STM 的实现原理。通过上面转账的例子，我们可以看到 STM 的使用特别简单，只需把业务相关的代码封装成可重入的函数传给 stm，而 STM 可自行处理事务相关的细节。

// 位于 clientv3/concurrency/stm.go:25type STM interface {// Get 返回键的值，并将该键插入 txn 的 read set 中。如果 Get 失败，它将以错误中止事务，没有返回Get(key ...string) string// Put 在 write set 中增加键值对Put(key, val string, opts ...v3.OpOption)// Rev 返回 read set 中某个键指定的版本号Rev(key string) int64// Del 删除某个键Del(key string)// commit 尝试提交事务到 etcd servercommit() *v3.TxnResponsereset()}

STM 是软件事务存储的接口。其中定义了 Get、Put、Rev、Del、commit、reset 接口方法。STM 的接口有两个实现类：stm 和 stmSerializable。具体选择哪一个，**由我们指定的隔离级别决定。

STM 对象在内部构造 txn 事务，业务函数转换成If-Then，自动提交事务以及处理失败重试等工作，直到事务执行成功。核心的NewSTM函数的实现如下所示：

// NewSTM initiates a new STM instance, using serializable snapshot isolation by default.func NewSTM(c *v3.Client, apply func(STM) error, so ...stmOption) (*v3.TxnResponse, error) {opts := &stmOptions{ctx: c.Ctx()}for _, f := range so {f(opts)}if len(opts.prefetch) != 0 {f := applyapply = func(s STM) error {s.Get(opts.prefetch...)return f(s)}}return runSTM(mkSTM(c, opts), apply)}

根据源码可以知道，NewSTM首先判断该事务是否存在预取的键值对，如果存在，会无条件地直接 apply 函数；否则会创建一个 stm，并运行 stm 事务。runSTM 代码如下所示：

// 位于 clientv3/concurrency/stm.go:140func runSTM(s STM, apply func(STM) error) (*v3.TxnResponse, error) {outc := make(chan stmResponse, 1)go func() {defer func() {if r := recover(); r != nil {e, ok := r.(stmError)if !ok {// 执行异常panic(r)}outc <- stmResponse{nil, e.err}}}()var out stmResponsefor {            // 重置 stms.reset()            // 执行事务操作，apply 函数if out.err = apply(s); out.err != nil {break}            // 提交事务if out.resp = s.commit(); out.resp != nil {break}}outc <- out}()r := <-outcreturn r.resp, r.err}

runSTM 函数首先重置了 stm，清空 STM 的读写缓存；接着执行事务操作，apply 应用函数；最后将事务提交。提交事务的实现如下：

// 位于 clientv3/concurrency/stm.go:265func (s *stm) commit() *v3.TxnResponse {   txnresp, err := s.client.Txn(s.ctx).If(s.conflicts()...).Then(s.wset.puts()...).Commit()   if err != nil {      panic(stmError{err})   }   if txnresp.Succeeded {      return txnresp   }   return nil}

上述 commit 的实现包含了我们前面所介绍的 etcd 事务语法。If 中封装了冲突检测条件，提交事务则是 etcd 的 Txn 将 wset 中的数据写入并提交的过程。

下面我们来看看 etcd 隔离级别以及在 STM 封装基础上如何实现事务。

4、etcd 事务隔离级别

数据库有如下几种事务隔离级别（Transaction Isolation Levels）：

未提交读（Read Uncommitted）：能够读取到其他事务中还未提交的数据，这可能会导致脏读的问题。
读已提交（Read Committed）：只能读取到已经提交的数据，即别的事务一提交，当前事务就能读取到被修改的数据，这可能导致不可重复读的问题。
可重复读（Repeated Read）：一个事务中，同一个读操作在事务的任意时刻都能得到同样的结果，其他事务的提交操作对本事务不会产生影响。
串行化（Serializable）：串行化的执行可能冲突的事务，即一个事务会阻塞其他事务。它通过牺牲并发能力来换取数据的安全，属于最高的隔离级别。

而 etcd clientv3 实现了四种事务模型，位于 clientv3/concurrency/stm.go 中，分别为 SerializableSnapshot、Serializable、RepeatableReads 和 ReadCommitted。

// 位于 clientv3/concurrency/stm.go:45const (// SerializableSnapshot provides serializable isolation and also checks// for write conflicts.SerializableSnapshot Isolation = iota// Serializable reads within the same transaction attempt return data// from the at the revision of the first read.Serializable// RepeatableReads reads within the same transaction attempt always// return the same data.RepeatableReads// ReadCommitted reads keys from any committed revision.ReadCommitted)// WithIsolation specifies the transaction isolation level.func WithIsolation(lvl Isolation) stmOption {return func(so *stmOptions) { so.iso = lvl }}

STM 的事务级别通过 stmOption 指定，默认就是 SerializableSnapshot。下面分别介绍这几种隔离级别。

构造 STM 的实现如下所示：

// 位于 clientv3/concurrency/stm.go:105func mkSTM(c *v3.Client, opts *stmOptions) STM {   switch opts.iso {   // 串行化快照   case SerializableSnapshot:      s := &stmSerializable{         stm:      stm{client: c, ctx: opts.ctx},         prefetch: make(map[string]*v3.GetResponse),      }      s.conflicts = func() []v3.Cmp {         return append(s.rset.cmps(), s.wset.cmps(s.rset.first()+1)...)      }      return s   // 串行化   case Serializable:      s := &stmSerializable{         stm:      stm{client: c, ctx: opts.ctx},         prefetch: make(map[string]*v3.GetResponse),      }      s.conflicts = func() []v3.Cmp { return s.rset.cmps() }      return s   // 可重复读      case RepeatableReads:      s := &stm{client: c, ctx: opts.ctx, getOpts: []v3.OpOption{v3.WithSerializable()}}      s.conflicts = func() []v3.Cmp { return s.rset.cmps() }      return s   // 已提交读   case ReadCommitted:      s := &stm{client: c, ctx: opts.ctx, getOpts: []v3.OpOption{v3.WithSerializable()}}      s.conflicts = func() []v3.Cmp { return nil }      return s   default:      panic("unsupported stm")   }}

该函数是根据隔离级别定义的。每一类隔离级别对应不同的冲突检测条件，存在读操作差异，因此我们需要搞清楚每一类隔离级别在这两方面的实现。

从构建 SMT 的实现代码可以知道，etcd 隔离级别与一般的数据库隔离级别的差异是没有未提交读的隔离级别，这是由于 etcd 的 kv 操作(包括 txn 事务内的多个 keys 操作)都是原子操作，所以你不可能读到未提交的修改。下面我们将从低到高分别介绍 etcd 事务隔离级别。

4.1 ReadCommitted

读提交是指，一个事务提交之后，它做的变更才会被其他事务看到。只允许获取已经提交的数据。比如事务 A 和事务 B 同时进行，事务 A 进行 +1 操作，此时，事务 B 无法看到这个数据项在事务A操作过程中的所有中间值，只能看到最终的 10。

由于 etcd 的 kv 操作(包括 txn 事务内的多个 keys 操作)都是原子操作，所以你不可能读到未提交的修改，ReadCommitted 是 etcd 中的最低事务级别。

由构造 STM 的源码可知，ReadCommitted 调用的是 stm 的实现。对于不一样的隔离级别，我们主要关注的就是读操作和提交时的冲突检测条件。而对于写操作，会先写进本地缓存，直到事务提交时才真正写到 etcd 里。

读操作

func (s *stm) Get(keys ...string) string {if wv := s.wset.get(keys...); wv != nil {return wv.val}return respToValue(s.fetch(keys...))}

从 etcd 读取 keys，就像普通的 kv 操作一样。第一次 Get 后，在事务中缓存，后续不再从 etcd 读取。

冲突检测条件

s.conflicts = func() []v3.Cmp { return nil }

没有任何冲突检测。

ReadCommitted 只需要确保自己读到的是别人已经提交的数据，由于 etcd 的 kv 操作都是原子操作，所以不可能读到未提交的修改。

4.2 RepeatableReads

可重复读是指，一个事务执行过程中看到的数据，总是跟这个事务在启动时看到的数据是一致的。多次读取同一个数据时，其值都和事务开始时刻是一致的，因此该事务级别解决了不可重复读取和脏读取的问题，但是有可能出现幻影数据。所谓幻影数据，在一个事务内多次查询某个符合查询条件的「记录数量」，如果出现前后两次查询到的记录数量不一样的情况，就意味着发生了「幻读」现象。

读操作

与 ReadCommitted 类似，用 readSet 缓存已经读过的数据，这样下次再读取相同数据的时候才能得到同样的结果，确保了可重复读。

冲突检测条件

s.conflicts = func() []v3.Cmp { return s.rset.cmps() }

在事务提交时，确保事务中 Get 的 keys 没有被改动过。因此使用 readSet 数据的 ModRevision 做冲突检测，确保本事务读到的数据都是最新的。

MySQL 事务“可重复读”是通过在事务第一次 select 时建立 readview，来确保事务中读到的是到这一刻为止的最新数据，忽略后面发生的更新。而这里每个 key 的 Get 是独立的（也可以说，每个 key 都是获取的当前值，没有 readview 的概念)，在事务提交时，如果这些 keys 没有变动过，那么事务就可以提交。

4.3 Serializable

串行化，顾名思义是对同一行记录，“写”会加“写锁”，“读”会加“读锁”。当出现读写锁冲突的时候，后访问的事务必须等前一个事务执行完成，才能继续执行。是最严格的事务隔离级别，它要求所有事务被串行执行。

读操作

func (s *stmSerializable) Get(keys ...string) string {   if wv := s.wset.get(keys...); wv != nil {      return wv.val   }   // 判断是否第一次读   firstRead := len(s.rset) == 0   for _, key := range keys {      if resp, ok := s.prefetch[key]; ok {         delete(s.prefetch, key)         s.rset[key] = resp      }   }   resp := s.stm.fetch(keys...)   if firstRead {      // 记录下第一次读的版本作为基准      s.getOpts = []v3.OpOption{         v3.WithRev(resp.Header.Revision),         v3.WithSerializable(),      }   }   return respToValue(resp)}

事务中第一次读操作完成时，保存当前版本号 Revision；后续其他读请求会带上这个版本号，获取指定 Revision 版本的数据。这确保了该事务所有的读操作读到的都是同一时刻的内容。

冲突检测条件

s.conflicts = func() []v3.Cmp { return s.rset.cmps() }

在事务提交时，需要检查事务中 Get 的 keys 是否被改动过，而 etcd 串行化的约束还不够，它缺少了验证事务要修改的 keys 这一步。下面的 SerializableSnapshot 事务增加了这个约束。

可见，这个约束比数据库串行化的约束要低，它没有验证事务要修改的 keys 是否被改动过，下面的 SerializableSnapshot 事务增加了这个约束。

4.4 SerializableSnapshot

SerializableSnapshot串行化快照隔离，提供可序列化的隔离，并检查写冲突。etcd 默认采用这种隔离级别，串行化快照隔离是最严格的隔离级别，可以避免幻影读。其读操作与冲突检测的过程如下。

读操作

与 Serializable 串行化读类似。事务中的第一个 Get 操作发生时，保存服务器返回的当前 Revision；后续对其他 keys 的 Get 操作，指定获取 Revision 版本的 value。

冲突检测条件

s.conflicts = func() []v3.Cmp {    return append(s.rset.cmps(), s.wset.cmps(s.rset.first()+1)...)}

在事务提交时，检查事务中 Get 的 keys 以及要修改的 keys 是否被改动过。

SerializableSnapshot 不仅确保了读取过的数据是最新的，同时也确保了要写入的数据同样没有被其他事务更改过，是隔离的最高级别。

通过上面的分析，我们清楚了如何使用 etcd 的 txn 事务，构建符合 ACID 语义的事务框架。如果这些语义不能满足你的业务需求，通过扩展 etcd 的官方 client sdk，写一个新 STM 事务类型即可。

需要强调的是，数据库事务是“锁/阻塞”模式，而 etcd 的 STM 事务是 “CAS/重试” 模式，这是有差别的。简单的说，数据库事务不会自己重试，而 STM 事务在发生冲突是会多次重试，必须要保证业务代码是可重试的，且必须有明确的失败条件(例如判断账户余额是否够转账)。

参考资料：

bilibili视频讲解etcd事务

ETCD 十分布式事务

彻底搞懂 etcd 系列文章（八）：etcd 事务 API

etcd系列之事务：etcd 中如何实现事务（上）？

事务：etcd 中如何实现事务（下）？

关键词：