01.事务

1.1 事务伪代码

Txn(context.TODO()).If(//如果以下判断条件成立
	Compare(Value(k1), "<", v1),
	Compare(Version(k1), "=", 2)
).Then(//则执行Then代码段
	OpPut(k2,v2), OpPut(k3,v3)
).Else(//否则执行Else代码段
	OpPut(k4,v4), OpPut(k5,v5)
).Commit()//最后提交事务

1.2 通过txn实现分布式锁

• ETCD中的txn通过简单的"If-Then-Else"实现了原子操作。

• 实现分布式锁主要分为三个步骤：

  • 第一步：上锁，包括创建租约、自动续约、在租约时间内去抢一个key
  • 第二步：抢到锁后执行业务逻辑，没有抢到退出
  • 第三步：释放租约

package main

import (
	"fmt"
	"github.com/coreos/etcd/clientv3"
	"time"
	"context"
)

func main()  {
	config := clientv3.Config{
		Endpoints:     []string{"127.0.0.1:2379"},
		DialTimeout:   10*time.Second,
	}
	cli,_ := clientv3.New(config)
	txnTest(cli)
}

func txnTest(cli *clientv3.Client)  {
	// 1. 上锁
	// 1.1 创建一个5s的租约
	lease := clientv3.NewLease(cli)
	leaseGrantResp, _ := lease.Grant(context.TODO(), 5)
	leaseId := leaseGrantResp.ID

	// 1.2 自动续约（创建一个可取消的租约，主要是为了退出的时候能够释放）
	ctx, cancelFunc := context.WithCancel(context.TODO())

	// 3. 释放租约
	defer cancelFunc()   // 取消的租约
	defer lease.Revoke(context.TODO(), leaseId)  // 回收对应的key

	keepRespChan, _ := lease.KeepAlive(ctx, leaseId)

	// 续约应答
	go func() {
		for {
			select {
			case keepResp := <- keepRespChan:
				if keepRespChan == nil {
					fmt.Println("租约已经失效了")
					goto END
				} else {    // 每秒会续租一次, 所以就会受到一次应答
					fmt.Println("收到自动续租应答:", keepResp.ID)
				}
			}
		}
	END:
	}()

	// 1.3 在租约时间内去抢锁（etcd里面的锁就是一个key）
	kv := clientv3.NewKV(cli)

	// 创建事物
	txn := kv.Txn(context.TODO())

	//if 不存在key， then 设置它, else 抢锁失败
	txn.If(clientv3.Compare(clientv3.CreateRevision("lock"), "=", 0)).
		Then(clientv3.OpPut("lock", "g", clientv3.WithLease(leaseId))).
		Else(clientv3.OpGet("lock"))

	// 提交事务
	txnResp, _ := txn.Commit()
	if !txnResp.Succeeded {
		fmt.Println("锁被占用:", string(txnResp.Responses[0].GetResponseRange().Kvs[0].Value))
		return
	}

	// 2. 抢到锁后执行业务逻辑，没有抢到退出
	fmt.Println("处理任务")
	time.Sleep(5 * time.Second)

	// 3. 释放锁，步骤在上面的defer，当defer租约关掉的时候，对应的key被回收了
}

/*
收到自动续租应答: 7587859680424737936
处理任务
收到自动续租应答: 7587859680424737936
收到自动续租应答: 7587859680424737936
 */