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

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

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

根据上面的实现，其实很好理解事务实现的逻辑。如果 If 冲突判断语句为真，对应返回值为 true，Then 中的语句将会被执行，否则执行 Else 中的逻辑。

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

etcd 事务的使用示例

我们来演示一个转账的过程，发送者向接收者发起转账事务。etcd 的事务基于乐观锁检测冲突并重试，检测冲突时使用了ModRevision进行校验，该字段表示某个 key 上一次被更改时，全局的版本是多少。因此，我们实现转账业务的流程如下所示：

在 etcd 中的实现代码如下所示：

func txnTransfer(etcd *v3.Client, sender, receiver string, amount uint) error {
	// 失败重试
	for {
		if ok, err := doTxn(etcd, sender, receiver, amount); err != nil {
			return err
		} else if ok {
			return nil
		}
	}
}
func doTxn(etcd *v3.Client, sender, receiver string, amount uint) (bool, error) {
	// 第一个事务，利用事务的原子性，同时获取发送和接收者的余额以及 ModRevision
	getresp, err := etcd.Txn(context.TODO()).Then(v3.OpGet(sender), v3.OpGet(receiver)).Commit()
	if err != nil {
		return false, err
	}
	senderKV := getresp.Responses[0].GetResponseRange().Kvs[0]
	receiverKV := getresp.Responses[1].GetResponseRange().Kvs[1]
	senderNum, receiverNum := toUInt64(senderKV.Value), toUInt64(receiverKV.Value)
	// 验证账户余额是否充足
	if senderNum < amount {
		return false, fmt.Errorf("资金不足")
	}
	// 发起转账事务，冲突判断 ModRevision 是否发生变化
	txn := etcd.Txn(context.TODO()).If(
		v3.Compare(v3.ModRevision(sender), "=", senderKV.ModRevision),
		v3.Compare(v3.ModRevision(receiver), "=", receiverKV.ModRevision))
	txn = txn.Then(
		v3.OpPut(sender, fromUint64(senderNum-amount)), // 更新发送者账户余额
		v3.OpPut(receiver, fromUint64(receiverNum-amount))) // 更新接收者账户余额
    resp, err := txn.Commit()         // 提交事务
	if err != nil {
		return false, err
	}
	return resp.Succeeded, nil
}

 

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

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

使用 STM 实现转账

为了简化 etcd 事务实现的过程，etcd clientv3 提供了 STM（Software Transactional Memory，软件事务内存），帮助我们自动处理这些烦琐的过程。使用 STM 优化之后的转账业务代码如下：

func txnStmTransfer(cli *v3.Client, from, to string, amount uint) 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(to, fromUint64(receiverNum + amount))
		stm.Put(from, fromUint64(senderNum - amount))
		return nil
	})
	return err
}

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

STM 实现细节

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

// 位于 clientv3/concurrency/stm.go:25
type 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 server
	commit() *v3.TxnResponse
	reset()
}

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 := apply
		apply = 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:140
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 {
					// 执行异常
					panic(r)
				}
				outc <- stmResponse{nil, e.err}
			}
		}()
		var out stmResponse
		for {
            // 重置 stm
			s.reset()
            // 执行事务操作，apply 函数
			if out.err = apply(s); out.err != nil {
				break
			}
            // 提交事务
			if out.resp = s.commit(); out.resp != nil {
				break
			}
		}
		outc <- out
	}()
	r := <-outc
	return r.resp, r.err
}

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

// 位于 clientv3/concurrency/stm.go:265
func (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 封装基础上如何实现事务。

etcd 事务隔离级别

数据库一般有以下几种事务隔离级别。

• 未提交读（Read Uncommitted）：能够读取到其他事务中还未提交的数据，这可能会导致脏读的问题。

• 读已提交（Read Committed）：只能读取到已经提交的数据，即别的事务一提交，当前事务就能读取到被修改的数据，这可能导致不可重复读的问题。

• 可重复读（Repeated Read）：一个事务中，同一个读操作在事务的任意时刻都能得到同样的结果，其他事务的提交操作对本事务不会产生影响。

• 串行化（Serializable）：串行化执行事务，即一个事务的执行会阻塞其他事务。该隔离级别通过牺牲并发能力换取数据的安全，属于最高的隔离级别。

etcd 的事务可以看作是一种“微事务”，在它之上，可以构建出各种隔离级别的事务。STM 的事务级别通过 stmOption 指定，位于 clientv3/concurrency/stm.go 中，分别为 SerializableSnapshot、Serializable、RepeatableReads 和 ReadCommitted。

构造 STM 的实现如下所示：

func 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 通过 MVCC 机制实现读写不阻塞，并解决脏读的问题。下面我们将从低到高分别介绍 etcd 事务隔离级别。

ReadCommitted 读已提交

ReadCommitted 是 etcd 中的最低事务级别。ReadCommitted 是指一个事务提交之后，它做的变更才会被其他事务看到，只允许客户端获取已经提交的数据。

由构造 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 操作都是原子操作，所以不可能读到未提交的修改。

RepeatableReads 可重复读

RepeatableReads 与 ReadCommitted 类似，调用的也是 stm 的实现。可重复读是指多次读取同一个数据时，其值都和事务开始时刻是一致的，因此可以实现可重复读。

• 读操作

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

• 冲突检测条件

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

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

可重复读隔离级别的场景中，每个 key 的 Get 是独立的。在事务提交时，如果这些 keys 没有变动过，那么事务就可以提交。

Serializable 串行读

串行化调用的实现类为 stmSerializable，当出现读写锁冲突的时候，后续事务必须等前一个事务执行完成，才能继续执行。这就相当于在事务开始时，对 etcd 做了一个快照，这样它读取到的数据就不会受到其他事务的影响，从而达到事务串行化（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 事务增加了这个约束。

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 的官方 Client SDK，写一个新 STM 事务类型即可。