当需要批量操作etcd多个key的时候，常常需要让多次操作形成一个原子性的效果，要么同时成功，要么同时失败。
etcd提供了事务的机制，可以实现多个key的原子性操作。
这里使用etcd提供的go clientV3进行操作
在事务中使用到了Op对象来执行操作，所以先介绍一下Op。
Op对象是为了简化用户操作而开发的，可以根据ClientV3提供的函数来创建对应类型的Op对象、
func OpDelete(key string, opts ...OpOption) Op //执行删除动作 func OpGet(key string, opts ...OpOption) Op //执行查询动作 func OpPut(key, val string, opts ...OpOption) Op //执行新增、修改动作 func OpTxn(cmps []Cmp, thenOps []Op, elseOps []Op) Op //执行事务
KV接口下的Do方法可以接收一个Op并执行对象定义的操作，返回结果OpResponse是各种类型返回结果的抽象，可以根据请求的类型获取对应的响应结果。
type KV interface {
Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error)
Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error)
Delete(ctx context.Context, key string, opts ...OpOption) (*DeleteResponse, error)
Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error)
Do(ctx context.Context, op Op) (OpResponse, error)
Txn(ctx context.Context) Txn
}
下面看一下使用Op操作etcd的示例
package main
import (
"log"
"time"
"github.com/coreos/etcd/clientv3"
"context"
"fmt"
)
func main() {
cli, err := clientv3.New(clientv3.Config{
Endpoints: []string{"127.0.0.1:2379"},
DialTimeout: time.Second * 5,
})
if err != nil {
log.Fatal(err)
}
defer cli.Close()
ops := []clientv3.Op{
clientv3.OpPut("aaa", "123"),
clientv3.OpGet("aaa"),
}
for _, op := range ops {
resp, err := cli.Do(context.TODO(), op)
if err != nil {
log.Fatal(err)
}
if op.IsPut() {
fmt.Println(resp.Put())
}
if op.IsGet() {
fmt.Println(resp.Get())
}
}
}
事务的API
etcd v3 为了解决多 key 的原子操作问题，提供的事务API是下面的格式：
client.Txn(ctx).If(cmp1, cmp2, ...).Then(op1, op2, ...,).Else(op1, op2, …)
事务 API 由 If 语句、Then 语句、Else 语句组成。使用的时候在If里面添加条件表达式，如果条件表达式全部通过，就执行Then中的操作，如果不通过则执行Else中的操作。
Then和Else中的操作就是上面介绍的Op对象，可以传入多个。
支持的条件判断有：
- func CreateRevision(key string) Cmp 根据创建版本进行判断
- func LeaseValue(key string) Cmp 根据租约Id进行判断
- func ModRevision(key string) Cmp 根据最后修改版本进行判断
- func Value(key string) Cmp 根据值进行判断
- func Version(key string) Cmp 根据当前key本身的版本进行判断
下面案例模仿userA向userB转账，先查询出各自资金和版本号，根据金额计算出转账后的各自资金，然后使用事务写回etcd，
事务中通过版本号判断如果双方账户没有被修改过事务就执行成功，否则不做修改。
直接使用命令行在etcd中造两条数据
$ etcdctl put userA 100 OK $ etcdctl put userB 200 OK $ etcdctl get --prefix user userA 100 userB 200
然后使用事务将userA值减100，userB增加100，根据版本号做判断，如果中途有别的线程操作则事务会失败。
package main
import (
"log"
"strconv"
"time"
"github.com/coreos/etcd/clientv3"
"context"
"fmt"
)
func main() {
cli, err := clientv3.New(clientv3.Config{
Endpoints: []string{"127.0.0.1:2379"},
DialTimeout: time.Second * 5,
})
if err != nil {
log.Fatal(err)
}
defer cli.Close()
type user struct {
key string
account int
modRevision int64
}
userA := user{key:"userA"}
userB := user{key:"userB"}
ctx, cancel := context.WithTimeout(context.Background(), 10* time.Second)
respA, err := cli.Get(ctx, userA.key)
cancel()
if err != nil {
fmt.Printf("获取数据失败 err:%vn", err)
return
}
if len(respA.Kvs) == 1 {
userA.modRevision = respA.Kvs[0].ModRevision
userA.account,_ = strconv.Atoi(string(respA.Kvs[0].Value))
}
ctx, cancel = context.WithTimeout(context.Background(), 10* time.Second)
respB, err := cli.Get(ctx, userB.key)
cancel()
if err != nil {
fmt.Printf("获取数据失败 err:%vn", err)
return
}
if len(respB.Kvs) == 1 {
userB.modRevision = respB.Kvs[0].ModRevision
userB.account,_ = strconv.Atoi(string(respB.Kvs[0].Value))
}
userA.account -= 100
userB.account += 100
txnReps,err := cli.Txn(context.TODO()).If(
clientv3.Compare(clientv3.ModRevision(userA.key), "=", userA.modRevision),
clientv3.Compare(clientv3.ModRevision(userB.key), "=", userB.modRevision),
).Then(
clientv3.OpPut(userA.key,strconv.Itoa(userA.account)),
clientv3.OpPut(userB.key,strconv.Itoa(userB.account)),
).Else(
//do something
).Commit()
if err != nil {
log.Printf("事务执行失败 err:%vn", err)
return
}
if !txnReps.Succeeded {
log.Println("事务失败")
return
}
log.Println("事务成功")
}
