etcd 分布式锁和事务

2020-12-25 | 129 浏览

etcd分布式锁和事务

1.概念

1.锁

单进程的系统中，存在多线程同时操作一个公共变量，此时需要加锁对变量进行同步操作，保证多线程的操作线性执行消除并发修改。解决的是单进程中的多线程并发问题。
思考，golang的channel和锁的关系

2.分布式锁

只要的应用场景是在集群模式的多个相同服务，可能会部署在不同机器上，解决进程间安全问题，防止多进程同时操作一个变量或者数据库。解决的是多进程的并发问题。

3.事务

解决一个会话过程中，上下文的修改对所有数据库表的操作要么全部成功，要不全部失败。
在互联网应用中，常常用补偿逻辑来替代事务回滚

2.分布式锁

无状态的服务容易水平扩展

有状态的数据修改操作，要时时考虑并发冲突

3.etcd支持事务

模式:

if compare

then op

else op

commit

4.etcd分布式锁实现原理

etcd:

利用租约在etcd集群中创建一个key，这个key有两种形态，存在和不存在，而这两种形态就是互斥量。

如果这个key不存在，那么线程创建key，成功则获取到锁，该key就为存在状态。

如果该key已经存在，那么线程就不能创建key，则获取锁失败。

5.分布式锁和事务demo

一致性算法 -> 支持事务 -> 分布式锁

demo.go

package main

import (
	"context"
	"fmt"
	"time"

	"go.etcd.io/etcd/clientv3"
)

func main() {
	var (
		config        clientv3.Config
		client        *clientv3.Client
		lease         clientv3.Lease
		leaseResp     *clientv3.LeaseGrantResponse
		leaseId       clientv3.LeaseID
		leaseRespChan <-chan *clientv3.LeaseKeepAliveResponse
		err           error
	)
	//客户端配置
	config = clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"},
		DialTimeout: 5 * time.Second,
	}
	//建立连接
	if client, err = clientv3.New(config); err != nil {
		fmt.Println(err)
		return
	}

	//上锁（创建租约，自动续租）
	lease = clientv3.NewLease(client)

	//设置一个ctx取消自动续租
	ctx, cancleFunc := context.WithCancel(context.TODO())

	//设置10秒租约（过期时间）
	if leaseResp, err = lease.Grant(context.TODO(), 10); err != nil {
		fmt.Println(err)
		return
	}
	//拿到租约id
	leaseId = leaseResp.ID

	//自动续租（不停地往管道中扔租约信息）
	if leaseRespChan, err = lease.KeepAlive(ctx, leaseId); err != nil {
		fmt.Println(err)

	}
	//启动一个协程去监听
	go listenLeaseChan(leaseRespChan)
	//业务处理
	kv := clientv3.NewKV(client)
	//创建事务
	txn := kv.Txn(context.TODO())
	txn.If(clientv3.Compare(clientv3.CreateRevision("/lock/resource1"), "=", 0)).
		Then(clientv3.OpPut("/lock/resource1", "true", clientv3.WithLease(leaseId))).
		Else(clientv3.OpGet("/lock/resource1")) //否则抢锁失败

	//提交事务
	if txtResp, err := txn.Commit(); err != nil {
		fmt.Println(err)
		return
	} else {
		//判断是否抢锁
		if !txtResp.Succeeded {
			fmt.Println("锁被占用：", string(txtResp.Responses[0].GetResponseRange().Kvs[0].Value))
			return
		}
	}

	fmt.Println("处理任务")

	//释放锁（停止续租，终止租约）
	defer cancleFunc()                          //函数退出取消自动续租
	defer lease.Revoke(context.TODO(), leaseId) //终止租约（去掉过期时间）

	time.Sleep(10 * time.Second)

}

func listenLeaseChan(leaseRespChan <-chan *clientv3.LeaseKeepAliveResponse) {
	var leaseKeepResp *clientv3.LeaseKeepAliveResponse
	for {
		select {
		case leaseKeepResp = <-leaseRespChan:
			if leaseKeepResp == nil {
				fmt.Println("租约失效了")
				goto END
			} else {
				fmt.Println(leaseKeepResp.ID)

			}

		}
	}
END:
}

6.运行demo

启动etcd

启动etcd-manager观察生成的key

在两个终端窗口运行，一个成功，一个失败

7.总结

分布式锁的简单实现，数据库乐观锁

取值，修改，提交

版本比较

简单修改 vs. 多操作事务

重试机制

超时机制

手工补偿

1.etcd
2.golang接入
3.分布式锁的场景

DevOps