分布式锁原理及实现

ゞ浴缸里的玫瑰 2022-05-17 03:14 199阅读 0赞

强烈推荐一个大神的人工智能的教程：http://www.captainbed.net/zhanghan

### 文章目录 ###

*  一、为什么需要锁?
 *  二、分布式锁方案比较
 *  三、Redis实现分布式锁
 *   *  \*\*1. 单个JVM下实现\*\*
     *  \*\*2.分布式下实现\*\*
 *  四、zookeeper实现分布式锁
 *   *  1.什么是zookeeper?
     *  2.zk的数据模型
     *  3.节点
     *  4.watch机制
     *  5.zk的一致性是如何保证的
     *  6.zk的锁实现原理
 *  五、数据库实现

![image][]

# 一、为什么需要锁? #

1.  多任务环境
2.  任务都需要对同一共享资源进行写操作
3.  对资源的访问是互斥的

# 二、分布式锁方案比较 #

![image][image 1]

# [三、Redis实现分布式锁][Redis] #

## **1. 单个JVM下实现** ##

代码块加Synchronized锁或者Reentrantlock锁

## **2.分布式下实现** ##

if（setnx（key，1） == 1）{
        expire（key，30）
        try {
            do something ......
        } finally {
            del（key）
        }
    }

**存在的问题:** 命令不是原子操作,可能出现还没有设置超时时间,服务宕机,导致锁永远存在  
**优化使用命令:**

set（key，1，30，NX）

**存在的问题:** 可能导致误删除,两个线程都来拿锁,A线程先获得锁之后,设置超时时间30s,在30s后线程还没执行到del命令行,这时锁就自己删掉了,这时候B线程就可以拿到锁,因此当A执行del的时候删掉的就是B的锁  
**优化:值为线程id,删除时判断是否是自己的key**

加锁：
    String threadId = Thread.currentThread().getId()
    set（key，threadId ，30，NX）
    
    解锁：
    if（threadId .equals(redisClient.get(key))）{
        del(key)
    }

**解决误删除问题:** 获得锁的线程开启一个守护线程,为这个锁续时,每20s执行一次,给key延时20s

# [四、zookeeper实现分布式锁][zookeeper] #

## 1.什么是zookeeper? ##

是一个分布式协调服务

## 2.zk的数据模型 ##

![image][image 2]

## 3.节点 ##

每一个节点都是一个Znode  
![image][image 3]  
data：  
Znode存储的数据信息。

ACL：  
记录Znode的访问权限，即哪些人或哪些IP可以访问本节点。

stat：  
包含Znode的各种元数据，比如事务ID、版本号、时间戳、大小等等。

child：  
当前节点的子节点引用，类似于二叉树的左孩子右孩子。

## 4.watch机制 ##

*  客户端访问服务端,服务端将在哈希表中存储被watch的znode的路径,对应value是watcher列表
 *  当znode被删除的时候,服务端会去哈希表中查找该znode对应的watcher列表,然后通知客户端,然后从哈希表中删除

## 5.zk的一致性是如何保证的 ##

通过ZAB协议保证,采用二阶段提交保证

*  客户端发送请求到任意Follower
 *  Follower将请求转发给Leader
 *  Leader采用二阶段提交，先发送Propose广播给Follower。
 *  Follower接到Propose消息，写入日志成功后，返回ACK消息给Leader。
 *  Leader接到半数以上ACK消息，返回成功给客户端，并且广播Commit请求给Follower。

## 6.zk的锁实现原理 ##

![image][image 4]

/**
     * zk实现分布式锁
     */
    @Service
    public class ZookeeperImproveLock implements Lock {
        private static final String LOCK_PATH="/LOCK";
        private static final String ZOOKEEPER_IP_PORT="localhost:2181";
        private ZkClient client = new ZkClient(ZOOKEEPER_IP_PORT,1000,1000,new SerializableSerializer());
        private CountDownLatch cdl;
        private String beforePath; //当前请求的节点前一个节点
        private String currentPath; //当前请求的节点
    
    //    判断有没有LOCK_PATH目录,没有则创建
        public ZookeeperImproveLock(){
            if(!this.client.exists(LOCK_PATH)){
                this.client.createPersistent(LOCK_PATH);
            }
        }
    
        @Override
        //非阻塞时加锁
        public boolean tryLock() {
            try {
    //        如果currentpath为空则为第一次尝试加锁,第一次加锁赋值currentpath
                if(currentPath==null||this.client.getChildren(LOCK_PATH).size() <=0){
        //            创建一个临时顺序节点
                    currentPath = this.client.createEphemeralSequential (LOCK_PATH + '/',"lock");
                    System.out.println("创建一个临时节点--->"+currentPath);
                }
                //获取所有临时节点并排序,临时节点名称为自增长的字符串,如:00000000400
                List<String> children = this.client.getChildren(LOCK_PATH);
                Collections.sort(children);
                System.out.println(children.toString());
                System.out.println(Thread.currentThread().getName()+ "get0--->"+ children.get(0));
                System.out.println(Thread.currentThread().getName()+"currentPath"+ currentPath);
                String realPath = LOCK_PATH+'/'+children.get(0);
                System.out.println(Thread.currentThread().getName()+ "real---"+ realPath);
                if(currentPath.equals(realPath)){
                    System.out.println(Thread.currentThread().getName()+"成功啦--->" + currentPath);
                    return true;
                }else{
                    //如果当前节点在所有节点中排名不是第一,则获取前面的节点名称,并赋值给beforepath
    //                int wz = 0;
    //                if(children.contains(currentPath)){
    //                    wz = children.indexOf(currentPath);
    //                }
                    int wz = Collections.binarySearch(children,currentPath.substring(6));
                    beforePath = LOCK_PATH + '/' + children.get(wz-1);
                    System.out.println(Thread.currentThread().getName()+"beforePath"+ beforePath);
                }
            } catch (RuntimeException e) {
                e.printStackTrace();
            }
            return false;
        }
    
        @Override
        public void lock() {
            if(!tryLock()){
                System.out.println("获取锁不成功--->" + currentPath);
                waitForLock();
                lock();
            }else{
                System.out.println("获得分布式锁--->"+currentPath);
            }
        }
    
        private void waitForLock() {
            IZkDataListener listener = new IZkDataListener() {
                @Override
                public void handleDataChange(String s, Object o) throws Exception {
    
                }
    
                @Override
                public void handleDataDeleted(String s) throws Exception {
                    if(cdl != null){
                        System.out.println("countdown" + currentPath);
                        cdl.countDown();
                    }
                }
            };
    
            //给排在前面的节点增加数据删除的watcher
            this.client.subscribeDataChanges(beforePath,listener);
    
            if(this.client.exists(beforePath)){
                cdl = new CountDownLatch(1);
                try {
                    cdl.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            this.client.unsubscribeDataChanges(beforePath,listener);
        }
    
        @Override
        public void lockInterruptibly() throws InterruptedException {
    
        }
    
    
    
        @Override
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
            return false;
        }
    
        @Override
        public void unlock() {
            //删除当前临时节点
            this.client.delete(currentPath);
            System.out.println("删除当前锁---->" + currentPath);
        }
    
        @Override
        public Condition newCondition() {
            return null;
        }
    }

# 五、数据库实现 #

1.  唯一索引
2.  行级锁

/**
     * mysql实现分布式锁
     */
    //@Service
    public class MysqlLock implements Lock {
        private static final int LOCK_NUM =1;
        @Resource(name = "lockMapper")
        private LockMapper lockMapper;
    
    //    阻塞式的实现
        public void lock() {
            if(!tryLock()){
                try {
                    Thread.currentThread().sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                lock();
            }
        }
    
    //    非阻塞式的实现
        public boolean tryLock() {
            try {
                lockMapper.insert(LOCK_NUM);
            } catch (Exception e) {
                return false;
            }
            return true;
        }
    
    //    解锁
        public void unlock() {
            lockMapper.deleteByPrimarykey(LOCK_NUM);
        }
    
        public Condition newCondition() {
            return null;
        }
    
        public void lockInterruptibly() throws InterruptedException {
    
        }
    
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
            return false;
        }
    }

[image]: /images/20220517/34b38cb6f7be43959dcc58ae96e610d5.png
[image 1]: /images/20220517/32e3d949c5b64352a279afc91dc2f905.png
[Redis]: https://mp.weixin.qq.com/s/8fdBKAyHZrfHmSajXT_dnA
[zookeeper]: https://mp.weixin.qq.com/s/Gs4rrF8wwRzF6EvyrF_o4A
[image 2]: /images/20220517/e053cd7796f9402fb9e2b3a3463ac702.png
[image 3]: /images/20220517/af9e18be21b44c15bd7cda9c4ad05988.png
[image 4]: /images/20220517/cb7522f51c3a46bcabd69426f4c7b92a.png