还不会使用分布式锁?从零开始基于 etcd 实现分布式锁

标签: 分布式锁 从零开始 etcd | 发表时间:2021-05-06 14:56 | 作者:aoho
出处:https://juejin.cn/backend

为什么需要分布式锁?

在单进程的系统中,当存在多个线程可以同时改变某个变量时,就需要对变量或代码块做同步,使其在修改这种变量时能够线性执行消除并发修改变量。而同步本质上通过锁来实现。为了实现多个线程在一个时刻同一个代码块只能有一个线程可执行,那么需要在某个地方做个标记,这个标记必须每个线程都能看到,当标记不存在时可以设置该标记,其余后续线程发现已经有标记了则等待拥有标记的线程结束同步代码块取消标记后再去尝试设置标记。

而在分布式环境下,数据一致性问题一直是难点。相比于单进程,分布式环境的情况更加复杂。分布式与单机环境最大的不同在于其不是多线程而是多进程。多线程由于可以共享堆内存,因此可以简单的采取内存作为标记存储位置。而进程之间甚至可能都不在同一台物理机上,因此需要将标记存储在一个所有进程都能看到的地方。

常见的是秒杀场景,订单服务部署了多个服务实例。如秒杀商品有 4 个,第一个用户购买 3 个,第二个用户购买 2 个,理想状态下第一个用户能购买成功,第二个用户提示购买失败,反之亦可。而实际可能出现的情况是,两个用户都得到库存为 4,第一个用户买到了 3 个,更新库存之前,第二个用户下了 2 个商品的订单,更新库存为 2,导致业务逻辑出错。

在上面的场景中,商品的库存是共享变量,面对高并发情形,需要保证对资源的访问互斥。在单机环境中,比如 Java 语言中其实提供了很多并发处理相关的 API,但是这些 API 在分布式场景中就无能为力了,由于分布式系统具备多线程和多进程的特点,且分布在不同机器中,synchronized 和 lock 关键字将失去原有锁的效果,。仅依赖这些语言自身提供的 API 并不能实现分布式锁的功能,因此需要我们想想其它方法实现分布式锁。

常见的锁方案如下:

  • 基于数据库实现分布式锁
  • 基于 Zookeeper 实现分布式锁
  • 基于缓存实现分布式锁,如 redis、etcd 等

下面我们简单介绍下这几种锁的实现,并重点介绍 etcd 实现锁的方法。

基于数据库的锁

基于数据库的锁实现也有两种方式,一是基于数据库表,另一种是基于数据库的排他锁。

基于数据库表的增删

基于数据库表增删是最简单的方式,首先创建一张锁的表主要包含下列字段:方法名,时间戳等字段。

具体使用的方法为:当需要锁住某个方法时,往该表中插入一条相关的记录。需要注意的是,方法名有唯一性约束。如果有多个请求同时提交到数据库的话,数据库会保证只有一个操作可以成功,那么我们就可以认为操作成功的那个线程获得了该方法的锁,可以执行方法体内容。执行完毕,需要删除该记录。

对于上述方案可以进行优化,如应用主从数据库,数据之间双向同步。一旦主库挂掉,将应用服务快速切换到从库上。除此之外还可以记录当前获得锁的机器的主机信息和线程信息,那么下次再获取锁的时候先查询数据库,如果当前机器的主机信息和线程信息在数据库可以查到的话,直接把锁分配给该线程,实现可重入锁。

基于数据库排他锁

我们还可以通过数据库的排他锁来实现分布式锁。基于 Mysql 的 InnoDB 引擎,可以使用以下方法来实现加锁操作:

   public void lock(){
    connection.setAutoCommit(false)
    int count = 0;
    while(count < 4){
        try{
            select * from lock where lock_name=xxx for update;
            if(结果不为空){
                //代表获取到锁
                return;
            }
        }catch(Exception e){

        }
        //为空或者抛异常的话都表示没有获取到锁
        sleep(1000);
        count++;
    }
    throw new LockException();
}
复制代码

在查询语句后面增加 for update,数据库会在查询过程中给数据库表增加排他锁。当某条记录被加上排他锁之后,其他线程无法再在该行记录上增加排他锁。其他没有获取到锁的就会阻塞在上述 select 语句上,可能的结果有 2 种,在超时之前获取到了锁,在超时之前仍未获取到锁。

获得排它锁的线程即可获得分布式锁,当获取到锁之后,可以执行业务逻辑,执行完业务之后释放锁。

基于数据库锁的总结

上面两种方式都是依赖数据库的一张表,一种是通过表中的记录的存在情况确定当前是否有锁存在,另外一种是通过数据库的排他锁来实现分布式锁。优点是直接借助现有的关系型数据库,简单且容易理解;缺点是操作数据库需要一定的开销,性能问题以及 SQL 执行超时的异常需要考虑。

基于 Zookeeper

基于 Zookeeper 的临时节点和顺序特性可以实现分布式锁。

申请对某个方法加锁时,在 Zookeeper 上与该方法对应的指定节点的目录下,生成一个唯一的临时有序节点。当需要获取锁时,只需要判断有序节点中该节点是否为序号最小的一个。业务逻辑执行完成释放锁,只需将这个临时节点删除即可。这种方式也可以避免由于服务宕机导致的锁无法释放,而产生的死锁问题。

Netflix 开源了一套 Zookeeper 客户端框架 curator,你可以自行去看一下具体使用方法。Curator 提供的 InterProcessMutex 是分布式锁的一种实现。acquire 方法获取锁,release 方法释放锁。另外,锁释放、阻塞锁、可重入锁等问题都可以有效解决。

关于阻塞锁的实现,客户端可以通过在 Zookeeper 中创建顺序节点,并且在节点上绑定监听器 Watch。一旦节点发生变化,Zookeeper 会通知客户端,客户端可以检查自己创建的节点是不是当前所有节点中序号最小的,如果是就获取到锁,便可以执行业务逻辑。

Zookeeper 实现的分布式锁也存在一些缺陷。在性能上可能不如基于缓存实现的分布式锁。因为每次在创建锁和释放锁的过程中,都要动态创建、销毁瞬时节点来实现锁功能。

此外,Zookeeper 中创建和删除节点只能通过 Leader 节点来执行,然后将数据同步到集群中的其他节点。分布式环境中难免存在网络抖动,导致客户端和 Zookeeper 集群之间的 session 连接中断,此时 Zookeeper 服务端以为客户端挂了,就会删除临时节点。其他客户端就可以获取到分布式锁了,导致了同时获取锁的不一致问题。

基于缓存实现分布式锁

相对于基于数据库实现分布式锁的方案来说,基于缓存来实现在性能方面会表现的更好一点,存取速度快很多。而且很多缓存是可以集群部署的,可以解决单点问题。基于缓存的锁有好几种,如memcached、redis、本文下面主要讲解基于 etcd 实现分布式锁。

通过 etcd txn 实现分布式锁

通过 etcd 实现分布式锁,同样需要满足一致性、互斥性和可靠性等要求。etcd 中的事务 txn、lease 租约以及 watch 监听特性,能够使得基于 etcd 实现上述要求的分布式锁。

思路分析

通过 etcd 的事务特性可以帮助我们实现一致性和互斥性。etcd 的事务特性,使用的 IF-Then-Else 语句,IF 语言判断 etcd 服务端是否存在指定的 key,即该 key 创建版本号 create_revision 是否为 0 来检查 key 是否已存在,因为该 key 已存在的话,它的 create_revision 版本号就不是 0。满足 IF 条件的情况下则使用 then 执行 put 操作,否则 else 语句返回抢锁失败的结果。当然,除了使用 key 是否创建成功作为 IF 的判断依据,还可以创建前缀相同的 key,比较这些 key 的 revision 来判断分布式锁应该属于哪个请求。

客户端请求在获取到分布式锁之后,如果发生异常,需要及时将锁给释放掉。因此需要租约,当我们申请分布式锁的时候需要指定租约时间。超过 lease 租期时间将会自动释放锁,保证了业务的可用性。是不是这样就够了呢?在执行业务逻辑时,如果客户端发起的是一个耗时的操作,操作未完成的请情况下,租约时间过期,导致其他请求获取到分布式锁,造成不一致。这种情况下则需要续租,即刷新租约,使得客户端能够和 etcd 服务端保持心跳。

具体实现

我们基于如上分析的思路,绘制出实现 etcd 分布式锁的流程图,如下所示:

image.png

基于 Go 语言实现的 etcd 分布式锁,测试代码如下所示:

   func TestLock(t *testing.T) {
// 客户端配置
config = clientv3.Config{
Endpoints:   []string{"localhost:2379"},
DialTimeout: 5 * time.Second,
}
// 建立连接
if client, err = clientv3.New(config); err != nil {
fmt.Println(err)
return
}

// 1. 上锁并创建租约
lease = clientv3.NewLease(client)

if leaseGrantResp, err = lease.Grant(context.TODO(), 5); err != nil {
panic(err)
}
leaseId = leaseGrantResp.ID

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

// 3. 释放租约
defer cancelFunc()
defer lease.Revoke(context.TODO(), leaseId)

if keepRespChan, err = lease.KeepAlive(ctx, leaseId); err != nil {
panic(err)
}
// 续约应答
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(client)

// 创建事务
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"))

// 提交事务
if txnResp, err = txn.Commit(); err != nil {
panic(err)
}

if !txnResp.Succeeded {
fmt.Println("锁被占用:", string(txnResp.Responses[0].GetResponseRange().Kvs[0].Value))
return
}

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

}
复制代码

预期的执行结果如下所示:

   === RUN   TestLock
处理任务
收到自动续租应答: 7587848943239472601
收到自动续租应答: 7587848943239472601
收到自动续租应答: 7587848943239472601
--- PASS: TestLock (5.10s)
PASS
复制代码

总得来说,如上关于 etcd 分布式锁的实现过程分为四个步骤:

  • 客户端初始化与建立连接;
  • 创建租约,自动续租;
  • 创建事务,获取锁;
  • 执行业务逻辑,最后释放锁。

创建租约的时候,需要创建一个可取消的租约,主要是为了退出的时候能够释放。释放锁对应的步骤,在上面的 defer 语句中。当 defer 租约关掉的时候,分布式锁对应的 key 就会被释放掉了。

小结

本文主要介绍了基于 etcd 实现分布式锁的案例。首先介绍了分布式锁产生的背景以及必要性,分布式架构不同于单体架构,涉及到多服务之间多个实例的调用,跨进程的情况下使用编程语言自带的并发原语没有办法实现数据的一致性,因此分布式锁出现,用来解决分布式环境中的资源互斥操作。接着介绍了基于数据库实现分布式锁的两种方式:数据表增删和数据库的排它锁。基于 Zookeeper 的临时节点和顺序特性也可以实现分布式锁,这两种方式或多或少存在性能和稳定性方面的缺陷。

接着本文重点介绍了基于 etcd 实现分布式锁的方案,根据 etcd 的特点,利用事务 txn、lease 租约以及 watch 监测实现分布式锁。

在我们上面的案例中,抢锁失败,客户端就直接返回了。那么当该锁被释放之后,或者持有锁的客户端出现了故障退出了,其他锁如何快速获取锁呢?所以上述代码可以基于 watch 监测特性进行改进,各位同学可以自行试试。

订阅最新文章,欢迎关注我的公众号

相关 [分布式锁 从零开始 etcd] 推荐:

还不会使用分布式锁?从零开始基于 etcd 实现分布式锁

- - 掘金 后端
在单进程的系统中,当存在多个线程可以同时改变某个变量时,就需要对变量或代码块做同步,使其在修改这种变量时能够线性执行消除并发修改变量. 为了实现多个线程在一个时刻同一个代码块只能有一个线程可执行,那么需要在某个地方做个标记,这个标记必须每个线程都能看到,当标记不存在时可以设置该标记,其余后续线程发现已经有标记了则等待拥有标记的线程结束同步代码块取消标记后再去尝试设置标记.

[转]consul VS zookeeper、etcd、doozerd

- - Xiao_Qiang_的专栏
  zookeeper、doozerd、etcd都有着相似的架构,这三者的服务节点都需要一个仲裁节点来操作,它们是强一致的,并提供各种操作原语. 应用程序可以通过客户端lib库来构建分布式的系统. 在一个单datacenter中,consul的server节点工作在一种简单的方式下,consul server需要一个仲裁操作,并提供强一致性.

[原]etcd系统简介

- - 用心做事
etcd是一个分布式可靠的键值存储系统. 它提供了与ZooKeeper相似的功能,但是使用Go语言编写而不是Java语言. Etcd使用Raft协调算法而不是ZooKeeper采用的Paxos算法. 在云计算方面,Go是一个大有前景的语言,被誉为云时代的C语言. 对比与ZooKeeper,etcd更轻量级,etc更加关注一下几点:.

从零开始搭建SRE

- - DockOne.io
【编者的话】Google在10年前创造了SRE这个工种. SRE,Site Reliability Engineering的缩写. 其中site是指Website,可以翻译为网站可靠性工程. 几年前资深Google SRE Chris Jones等人联合撰写了《Google SRE: How Google runs production systems》,首次向外界解密了Google的生产环境以及整个SRE的方法论.

服务发现:Zookeeper vs etcd vs Consul

- - 企业架构 - ITeye博客
服务发现:Zookeeper vs etcd vs Consul. 【编者的话】本文对比了Zookeeper、etcd和Consul三种服务发现工具,探讨了最佳的服务发现解决方案,仅供参考. 如果使用预定义的端口,服务越多,发生冲突的可能性越大,毕竟,不可能有两个服务监听同一个端口. 管理一个拥挤的比方说被几百个服务所使用的所有端口的列表,本身就是一个挑战,添加到该列表后,这些服务需要的数据库和数量会日益增多.

从零开始做工作室!

- l - 果壳网 guokr.com - 果壳网
DIYer:dwj300 制作时间:1周 制作难度:★★★★☆ GEEK指数:★★★★☆. 在“被没有足够工作空间”、“找不到工具和材料”、“花了太多时间找东西”弄到抓狂以后,我们决定清理地下杂物室,做个工作间. 在施工过程中我们学到了很多,希望这些经验会对想建设或者拓展工作空间的人们有帮助.

从零开始的Spring Session(二)

- - 程序猿DD | 博客
上一篇文章介绍了一些Session和Cookie的基础知识,这篇文章开始正式介绍Spring Session是如何对传统的Session进行改造的. 官网这么介绍Spring Session:. 其具体的特性非常之多,具体的内容可以从文档中了解到,笔者做一点自己的总结,Spring Session的特性包括但不限于以下:.

解读Google分布式锁服务

- XiaoHui - NoSQLfan
在2010年4月,Google的网页索引更新实现了实时更新,在今年的OSDI大会上,Google首次公布了有关这一技术的论文. 在此之前,Google的索引更新,采用的的批处理的方式(map/reduce),也就是当增量数据达到一定规模之后,把增量数据和全量索引库Join,得到最新的索引数据. 采用新的索引更新系统之后,数据的生命周期缩短了50%,所谓的数据生命周期是指,数据从网页上爬下来,到展现在搜索结果中这段时间间隔,但是正如Google所强调的,这一系统仅仅是为增量更新所建立的,并没有取代map/reduce的批量作业处理模式.

如何用redis实现分布式锁

- - CSDN博客数据库推荐文章
redis作为一个强大的key/value数据库,其实还可以用来实现轻量级的分布式锁. 最早官方在 SETNX命令页给了一个实现:. 不过这个方案有漏洞,就是release lock用的DEL命令不支持cas删除(delete if current value equals old value),这样忽略race condition将会出现问题:.

分布式锁看这篇就够了

- - zhisheng的博客
转载请务必注明原创地址为: http://www.54tianzhisheng.cn/2018/04/24/Distributed_lock/. 在单进程的系统中,当存在多个线程可以同时改变某个变量(可变共享变量)时,就需要对变量或代码块做同步,使其在修改这种变量时能够线性执行消除并发修改变量. 而同步的本质是通过锁来实现的.