写点什么

分布式锁的实现方案

用户头像
360技术
关注
发布于: 2021 年 03 月 24 日
分布式锁的实现方案

奇技 · 指南

Q:什么是分布式锁?

A:当多个进程在同一个系统中,用分布式锁控制多个进程对资源的访问。


分布式锁应用场景

  1. 传统的单体应用单机部署情况下,可以使用 java 并发处理相关的 API 进行互斥控制。

  2. 分布式系统后由于多线程,多进程分布在不同机器上,使单机部署情况下的并发控制锁策略失效,为了解决跨 JVM 互斥机制来控制共享资源的访问,这就是分布式锁的来源;分布式锁应用场景大都是高并发、大流量场景。


分布式锁实现

  • 基于 redis 的分布式锁


1. 加锁机制:根据 hash 节点选择一个客户端执行 lua 脚本

2. 锁互斥机制:再来一个客户端执行同样的 lua 脚本会提示已经存在锁,然后进入循环一直尝试加锁

3. 可重入机制

4. watch dog 自动延期机制

5. 释放锁机制



测试用例

单机


private RedissonClient getClient(){        Config config = new Config();        config.useSingleServer().setAddress("redis://127.0.0.1:6379");//.setPassword("");//.setConnectionMinimumIdleSize(10).setConnectionPoolSize(10);//.setConnectionPoolSize();//172.16.10.164        RedissonClient redissonClient = Redisson.create(config);        return redissonClient;    }    private ExecutorService executorService = Executors.newCachedThreadPool();    @Test    public void test() throws Exception {        int[] count = {0};        for (int i = 0; i < 10; i++) {            RedissonClient client = getClient();            final RedisLock redisLock = new RedisLock(client,"lock_key");            executorService.submit(() -> {                try {                    redisLock.lock();                    count[0]++;                } catch (Exception e) {                    e.printStackTrace();                } finally {                    try {                        redisLock.unlock();                    } catch (Exception e) {                        e.printStackTrace();                    }                }            });        }        executorService.shutdown();        executorService.awaitTermination(1, TimeUnit.HOURS);        System.out.println(count[0]);    }
复制代码


RedLock

public static RLock create (String url, String key){        Config config = new Config();        config.useSingleServer().setAddress(url);        RedissonClient redissonClient = Redisson.create(config);        return redissonClient.getLock(key);    }
RedissonRedLock redissonRedLock = new RedissonRedLock( create("redis://redis://127.0.0.1:6379","lock_key1"), create("redis://redis://127.0.0.1:6380","lock_key2"), create("redis://redis://127.0.0.1:6381","lock_key3")); RedisRedLock redLock = new RedisRedLock(redissonRedLock);
private ExecutorService executorService = Executors.newCachedThreadPool();
@Test public void test() throws Exception { int[] count = {0}; for (int i = 0; i < 2; i++) { executorService.submit(() -> { try { redLock.lock(); count[0]++; } catch (Exception e) { e.printStackTrace(); } finally { try { redLock.unlock(); } catch (Exception e) { e.printStackTrace(); } } }); } executorService.shutdown(); executorService.awaitTermination(1, TimeUnit.HOURS); System.out.println(count[0]); }
复制代码


基于 ETCD 实现分布式锁分析

ETCD 分布式锁的实现

1. Lease 机制:租约机制(TTL,Time To Live),Etcd 可以为存储的 key-value 对设置租约,当租约到期,key-value 将失效删除;同时也支持续约,通过客户端可以在租约到期之前续约,以避免 key-value 对过期失效。Lease 机制可以保证分布式锁的安全性,为锁对应的 key 配置租约,即使锁的持有者因故障而不能主动释放锁,锁也会因租约到期而自动释放。


2. Revision 机制:每个 key 带有一个 Revision 号,每进行一次事务加一,它是全局唯一的,通过 Revision 的大小就可以知道进行写操作的顺序。在实现分布式锁时,多个客户端同时抢锁,根据 Revision 号大小依次获得锁,可以避免 “羊群效应” ,实现公平锁。


3. Prefix 机制:即前缀机制。例如,一个名为 /etcdlock 的锁,两个争抢它的客户端进行写操作,实际写入的 key 分别为:key1="/etcdlock/UUID1",key2="/etcdlock/UUID2",其中,UUID 表示全局唯一的 ID,确保两个 key 的唯一性。写操作都会成功,但返回的 Revision 不一样,那么,如何判断谁获得了锁呢?通过前缀 /etcdlock 查询,返回包含两个 key-value 对的的 KeyValue 列表,同时也包含它们的 Revision,通过 Revision 大小,客户端可以判断自己是否获得锁。


4. Watch 机制:即监听机制,Watch 机制支持 Watch 某个固定的 key,也支持 Watch 一个范围(前缀机制),当被 Watch 的 key 或范围发生变化,客户端将收到通知;在实现分布式锁时,如果抢锁失败,可通过 Prefix 机制返回的 KeyValue 列表获得 Revision 比自己小且相差最小的 key(称为 pre-key),对 pre-key 进行监听,因为只有它释放锁,自己才能获得锁,如果 Watch 到 pre-key 的 DELETE 事件则说明 pre-key 已经释放,自己已经持有锁。



  • 基于 ETCD 分布式锁

步骤 1:建立连接

客户端连接 Etcd,以 /etcd/lock 为前缀创建全局唯一的 key,假设第一个客户端对应的 key="/etcd/lock/UUID1",第二个为 key="/etcd/lock/UUID2";客户端分别为自己的 key 创建租约 - Lease,租约的长度根据业务耗时确定;


步骤 2:创建定时任务作为租约的“心跳”

当一个客户端持有锁期间,其它客户端只能等待,为了避免等待期间租约失效,客户端需创建一个定时任务作为“心跳”进行续约。此外,如果持有锁期间客户端崩溃,心跳停止,key 将因租约到期而被删除,从而锁释放,避免死锁。


步骤 3:客户端将自己全局唯一的 key 写入 Etcd

执行 put 操作,将步骤 1 中创建的 key 绑定租约写入 Etcd,根据 Etcd 的 Revision 机制,假设两个客户端 put 操作返回的 Revision 分别为 1、2,客户端需记录 Revision 用以接下来判断自己是否获得锁


步骤 4:客户端判断是否获得锁

客户端以前缀 /etcd/lock/ 读取 keyValue 列表,判断自己 key 的 Revision 是否为当前列表中最小的,如果是则认为获得锁;否则监听列表中前一个 Revision 比自己小的 key 的删除事件,一旦监听到删除事件或者因租约失效而删除的事件,则自己获得锁。


步骤 5:执行业务

获得锁后,操作共享资源,执行业务代码


步骤 6:释放锁

完成业务流程后,删除对应的 key 释放锁

eg:


public class EtcdDistributeLock extends AbstractLock{
private Client client; private Lock lockClient; private Lease leaseClient; private String lockKey; private String lockPath; /** 锁的次数 */ private AtomicInteger lockCount; /** 租约有效期,防止客户端崩溃,可在租约到期后自动释放锁;另一方面,正常执行过程中,会自动进行续租,单位 ns */ private Long leaseTTL; /** 续约锁租期的定时任务,初次启动延迟,单位默认为 s,默认为1s,可根据业务定制设置*/ private Long initialDelay = 0L; /** 定时任务线程池类 */ ScheduledExecutorService service = null; /** 保存线程与锁对象的映射,锁对象包含重入次数,重入次数的最大限制为Int的最大值 */ private final ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap();
public EtcdDistributeLock(){}
public EtcdDistributeLock(Client client, String lockKey, long leaseTTL,TimeUnit unit){ this.client = client; lockClient = client.getLockClient(); leaseClient = client.getLeaseClient(); this.lockKey = lockKey; // 转纳秒 this.leaseTTL = unit.toNanos(leaseTTL); service = Executors.newSingleThreadScheduledExecutor(); }

@Override public void lock() { // 检查重入性 Thread currentThread = Thread.currentThread(); LockData oldLockData = threadData.get(currentThread); if (oldLockData != null && oldLockData.isLockSuccess()) { // re-entering int lockCount = oldLockData.lockCount.incrementAndGet(); if(lockCount < 0 ){ throw new Error("超出可重入次数限制"); } return; }
// 记录租约 ID Long leaseId = 0L; try{ leaseId = leaseClient.grant(TimeUnit.NANOSECONDS.toSeconds(leaseTTL)).get().getID(); // 续租心跳周期 long period = leaseTTL - leaseTTL / 5; // 启动定时任务续约 service.scheduleAtFixedRate(new EtcdDistributeLock.KeepAliveRunnable(leaseClient, leaseId), initialDelay,period,TimeUnit.NANOSECONDS); LockResponse lockResponse = lockClient.lock(ByteSequence.from(lockKey.getBytes()), leaseId).get(); if(lockResponse != null){ lockPath = lockResponse.getKey().toString(Charset.forName("utf-8")); log.info("获取锁成功,锁路径:{},线程:{}",lockPath,currentThread.getName()); } }catch (InterruptedException | ExecutionException e){ log.error("获取锁失败",e); return; } // 获取锁成功,锁对象设置 LockData newLockData = new LockData(currentThread, lockKey); newLockData.setLeaseId(leaseId); newLockData.setService(service); threadData.put(currentThread, newLockData); newLockData.setLockSuccess(true); }
@Override public void lockInterruptibly() throws InterruptedException { super.lockInterruptibly(); }
@Override public boolean tryLock() { return super.tryLock(); }
@Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { return super.tryLock(time,unit); }

@Override public void unlock() { Thread currentThread = Thread.currentThread(); LockData lockData = threadData.get(currentThread); if (lockData == null){ throw new IllegalMonitorStateException("You do not own the lock: " + lockKey); } int newLockCount = lockData.lockCount.decrementAndGet(); if ( newLockCount > 0 ) { return; } if ( newLockCount < 0 ) { throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + lockKey); } try { // 释放锁 if(lockPath != null){ lockClient.unlock(ByteSequence.from(lockPath.getBytes())).get(); } if(lockData != null){ // 关闭定时任务 lockData.getService().shutdown(); // 删除租约 if (lockData.getLeaseId() != 0L) { leaseClient.revoke(lockData.getLeaseId()); } } } catch (InterruptedException | ExecutionException e) { log.error("解锁失败",e); }finally { // 移除当前线程资源 threadData.remove(currentThread); } }

@Override public Condition newCondition() { return super.newCondition(); }
/** * 心跳续约线程类 */ public static class KeepAliveRunnable implements Runnable { private Lease leaseClient; private long leaseId;
public KeepAliveRunnable(Lease leaseClient, long leaseId) { this.leaseClient = leaseClient; this.leaseId = leaseId; }
@Override public void run() { // 对该leaseid进行一次续约 leaseClient.keepAliveOnce(leaseId); } }
复制代码


public class EtcdLockTest {    private Client client;    private String key = "/etcd/lock";    private static final String server = "http://xxxx:xxxx";    private ExecutorService executorService = Executors.newFixedThreadPool(10000);
@Before public void before() throws Exception { initEtcdClient(); }
private void initEtcdClient(){ client = Client.builder().endpoints(server).build(); }
@Test public void testEtcdDistributeLock() throws InterruptedException { int[] count = {0}; for (int i = 0; i < 100; i++) { executorService.submit(() -> { final EtcdDistributeLock lock = new EtcdDistributeLock(client, key,20,TimeUnit.SECONDS); try { lock.lock(); count[0]++; } catch (Exception e) { e.printStackTrace(); } finally { try { lock.unlock(); } catch (Exception e) { e.printStackTrace(); } } }); } executorService.shutdown(); executorService.awaitTermination(1, TimeUnit.HOURS); System.err.println("执行结果: " + count[0]); }}
复制代码


  • 基于 Zookeeper 分布式锁

实现原理

1. 启动客户端,确认链接到了服务器

2. 多个客户端并发的在特定路径下创建临时性顺序节点

3. 客户端判断自己的创建的顺序节点是否是最小的,如果是最小的,则获取锁成功

4. 第三步若判定失败,则采用 zk 的 watch 机制监听自己的前一个顺序节点,等待前一个节点的删除(放锁)事件,再开始第三步判定



zookeeper 作为高性能分布式协调框架,可以把其看做一个文件系统,其中有节点的概念,并且分为 4 种:1.持久性节点 2.持久性顺序节点 3.临时性节点 4.临时性顺序节点。


分布式锁的实现主要思路就是:监控其他客户端的状态,来判断自己是否可以获得锁。

采用临时性顺序节点的原因:


1. zk 服务器维护了客户端的会话有效性,当会话失效的时候,其会话所创建的临时性节点都会被删除,通过这一特点,可以通过 watch 临时节点来监控其他客户端的情况,方便自己做出相应动作。


2. 因为 zk 对写操作是顺序性的,所以并发创建的顺序节点会有一个唯一确定的序号,当前锁是公平锁的一种实现,所以依靠这种顺序性可以很好的解释—节点序列小的获取到锁并且可以采用 watch 自己的前一个节点来避免惊群现象(这样 watch 事件的传播是线性的)。

eg:


public class ZKLock extends AbstractLock {
/** * 1.Connect to zk */ private CuratorFramework client;
private InterProcessLock lock ;

public ZKLock(String zkAddress,String lockPath) { // 1.Connect to zk client = CuratorFrameworkFactory.newClient( zkAddress, new RetryNTimes(5, 5000) ); client.start(); if(client.getState() == CuratorFrameworkState.STARTED){ log.info("zk client start successfully!"); log.info("zkAddress:{},lockPath:{}",zkAddress,lockPath); }else{ throw new RuntimeException("客户端启动失败。。。"); } this.lock = defaultLock(lockPath); }
private InterProcessLock defaultLock(String lockPath ){ return new InterProcessMutex(client, lockPath); } @Override public void lock() { try { this.lock.acquire(); } catch (Exception e) { throw new RuntimeException(e); } }
@Override public boolean tryLock() { boolean flag ; try { flag=this.lock.acquire(0,TimeUnit.SECONDS); } catch (Exception e) { throw new RuntimeException(e); } return flag; }
@Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { boolean flag ; try { flag=this.lock.acquire(time,unit); } catch (Exception e) { throw new RuntimeException(e); } return flag; }
@Override public void unlock() { try { this.lock.release(); } catch (Exception e) { throw new RuntimeException(e); } }
}
复制代码


private ExecutorService executorService = Executors.newCachedThreadPool();

@Test public void testLock() throws Exception{ ZKLock zkLock = new ZKLock("xxxx:xxxx","/lockPath"); int[] num = {0}; long start = System.currentTimeMillis(); for(int i=0;i<200;i++){ executorService.submit(()->{ try { zkLock.lock(); num[0]++; } catch (Exception e){ throw new RuntimeException(e); } finally { zkLock.unlock(); } });
} executorService.shutdown(); executorService.awaitTermination(1, TimeUnit.HOURS); log.info("耗时:{}",System.currentTimeMillis()-start); System.out.println(num[0]); }
复制代码


总结

  1. redis 的分布式锁中 redisson 一般为单实例,当单实例不可用时,会阻塞业务流程。主从方式、主从数据异步,会存在锁失效的问题。RedLock 一般要求至少 3 台以上的 redis 主从实例,维护成本相对来说比较高。


2. ZK 锁具备高可用、可重入、阻塞锁特性,可解决失效死锁问题。但是因为需要频繁的创建和删除节点,性能上不如 Redis 方式。


3. ETCD 分布式锁的实现原理与 zk 锁类似,但是 ETCD 分布式锁更加可靠强大。其 Lease 功能保证分布式锁的安全性;watch 功能支持监听某个固定的 key,也支持 watch 一个范围的 key(前缀机制);revision 功能可通过 Revision 的大小就可以知道进行写操作的顺序。可以避免 “羊群效应” (也称 “惊群效应”),实现公平锁。前缀机制与 watch 功能配合使用解决了死锁问题。总之 ETCD 的灵感来源于 Zookeeper,但实现的时候做了很多的改进,如:高负载下的稳定读写、数据模型的多版本并发控制、稳定的 watch 功能,通知订阅者监听值得变化、可以容忍脑裂现场的发生、客户端的协议使用 gRPC 协议,支持 go、c++、java 等。


原文链接:分布式锁的实现方案


用户头像

360技术

关注

还未添加个人签名 2020.12.09 加入

还未添加个人简介

评论

发布
暂无评论
分布式锁的实现方案