分布式锁之zookeeper、curator实现与手写实现

柔光的暖阳◎ 2022-12-08 05:13 119阅读 0赞

# 分布式锁之zookeeper #

这里使用的zookeeper版本为`apache-zookeeper-3.6.2-bin`，下载地址：https://www.apache.org/dyn/closer.lua/zookeeper/zookeeper-3.6.2/apache-zookeeper-3.6.2-bin.tar.gz。

## curator中分布式锁的使用 ##

pom.xml中引入maven依赖：

<dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>5.1.0</version>
    </dependency>

具体使用如下：

package com.morris.distribute.lock.zookeeper.curator;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.framework.recipes.locks.InterProcessMutex;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    import org.junit.Before;
    import org.junit.Test;
    
    import java.util.concurrent.CyclicBarrier;
    import java.util.concurrent.TimeUnit;
    import java.util.stream.IntStream;
    
    @Slf4j
    public class Demo { 
    
        private CuratorFramework client;
    
        private String root = "/updateStatus";
    
        @Before
        public void init() { 
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
            // CuratorFramework client = CuratorFrameworkFactory.newClient("192.169.28.51:2181,192.169.28.52:2181,192.169.28.53:2181/testLock", retryPolicy);
            CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", retryPolicy);
            this.client = client;
            client.start();
            try { 
                if(null == client.checkExists().forPath(root)) { 
                    client.create().forPath(root, "myData".getBytes());
                }
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    
        @Test
        public void test() throws InterruptedException { 
    
            CyclicBarrier cyclicBarrier = new CyclicBarrier(10);
            IntStream.rangeClosed(1, 10).forEach((i) -> new Thread(() -> { 
    
                InterProcessMutex lock = new InterProcessMutex(client, root);
    
                try { 
                    cyclicBarrier.await();
                    lock.acquire(); // 加锁
                } catch (Exception e) { 
                    e.printStackTrace();
                }
    
                try { 
                    log.info("get lock");
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    try { 
                        lock.release(); // 释放锁
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            }, "t" + i).start());
    
            TimeUnit.SECONDS.sleep(100); // 保证上面10个线程跑起来
        }
    }

运行结果如下：

2020-09-17 17:01:00,920  INFO [t9] (Demo.java:55) - get lock
    2020-09-17 17:01:06,158  INFO [t10] (Demo.java:55) - get lock
    2020-09-17 17:01:11,195  INFO [t8] (Demo.java:55) - get lock
    2020-09-17 17:01:16,219  INFO [t2] (Demo.java:55) - get lock
    2020-09-17 17:01:21,253  INFO [t5] (Demo.java:55) - get lock
    2020-09-17 17:01:26,286  INFO [t1] (Demo.java:55) - get lock
    2020-09-17 17:01:31,319  INFO [t3] (Demo.java:55) - get lock
    2020-09-17 17:01:36,361  INFO [t7] (Demo.java:55) - get lock
    2020-09-17 17:01:41,394  INFO [t4] (Demo.java:55) - get lock
    2020-09-17 17:01:46,427  INFO [t6] (Demo.java:55) - get lock

可以看到10个线程串行执行。

同时在/updateStatus节点下创建了10个临时顺序节点：

[zk: localhost:2181(CONNECTED) 45] ls /updateStatus
    [_c_083fdc1a-a3ca-4bd5-a98e-093c0b863213-lock-0000000104, _c_466cbfa2-3b91-44eb-
    a5c1-38ae0f1a170d-lock-0000000106, _c_610796d3-95c7-4592-8005-d49fa718b635-lock-
    0000000107, _c_7f3a8efe-49c7-40c2-aad7-7d97aa8f0a00-lock-0000000103, _c_c71c9e00
    -59e5-4295-a7bc-e9440d5b117f-lock-0000000100, _c_ca0f24ba-92ff-4faf-a1b3-5688cd2
    ff53c-lock-0000000108, _c_d22ea891-9509-4d00-ad31-4061066a452c-lock-0000000102,
    _c_da8a987e-0bd8-4931-aec9-9796dd8a4cd5-lock-0000000109, _c_e4a508d9-257c-4c29-a
    174-cb344d961115-lock-0000000105, _c_fa33a67b-3d7d-4d8c-a152-cc01db47b293-lock-0
    000000101]

## 手写Zookeeper分布式锁 ##

下面使用Zookeeper的客户端zookeeper来实现Zookeeper分布式锁，由于项目中引用了`curator-recipes`，会自动导入如下依赖：

<dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.6.0</version>
    </dependency>

> 注意zookeeper的版本与api的版本保持一致，否则会出现一些奇怪的现象。

### 怎么样才算加锁成功？ ###

准备一个目录节点，例如`/zookeeperLock`，所有有关分布式锁的节点都将在这个目录下创建。然后每一把锁都对应`/zookeeperLock`中一个子节点，例如`/zookeeperLock/updateStatus1`是一把更新订单状态的锁，各个客户端在可以在这个锁目录节点下面创建子节点，类型为临时顺序节点，谁在这个目录下创建的子节点的序号最小，谁就获得锁。

假如在锁目录节点`/zookeeperLock/updateStatus1`（一般用与业务相关的名词命名）下，有10个客户端来竞争锁，那么将会在这个目录下会创建10个临时顺序节点，类似如下：

lock0000000001
    lock0000000002
    lock0000000003
    lock0000000004
    lock0000000005
    lock0000000006
    lock0000000007
    lock0000000008
    lock0000000009
    lock0000000010

此时获取锁的客户端为创建节点为`/lock0000000001`的客户端。

### 客户端挂了怎么办？ ###

采用临时顺序节点，节点的生命周期为session级别，如果客户端挂了，那么这个节点也会自动删除，不会造成死锁。

### 前面一个客户解锁了，怎么通知下一个客户端 ###

后面的客户端监听（watch）自己前面一个节点的删除事件，如果前面的节点被删除了（有可能是释放锁主动删除，也有可能是客户端挂了被动删除），那么后面的节点监听到事件后，再去遍历整个目录节点，看有没有比自己小的节点，如果没有就获取锁成功，如果有就继续监听比自己小1的节点的删除事件。

如果前面的节点被删除了，后面的节点不能直接获取锁，因为有可能目前A、B、C三个客户端都在等待获取，如果B挂了，那么C直接获取锁就不对了，必须再去看看有没有比自己小的节点。

### zookeeper分布式锁的实现 ###

package com.morris.distribute.lock.zookeeper.my;
    
    import org.apache.zookeeper.*;
    import org.apache.zookeeper.data.Stat;
    
    import java.util.Collections;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    
    public class ZookeeperLock implements Watcher, AsyncCallback.StringCallback, AsyncCallback.ChildrenCallback, AsyncCallback.StatCallback { 
    
        private ZooKeeper zooKeeper;
    
        public ZookeeperLock(ZooKeeper zooKeeper) { 
            this.zooKeeper = zooKeeper;
        }
    
        private CountDownLatch countDownLatch = new CountDownLatch(1);
    
        private String pathName;
    
        private String key;
    
        public void lock(String key) { 
            try { 
                this.key = key;
                byte[] value = Thread.currentThread().getName().getBytes();
                zooKeeper.create(key + "/lock", value, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, this, "lock");
                countDownLatch.await();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        // create的回调
        @Override
        public void processResult(int rc, String path, Object ctx, String name) { 
            this.pathName = name;
            // 获取根节点下所有的临时顺序节点
            zooKeeper.getChildren(key, false, this, name);
        }
    
    
        // getChildren的回调
        @Override
        public void processResult(int rc, String path, Object ctx, List<String> children) { 
    
            Collections.sort(children);
    
            int index = children.indexOf(pathName.substring(pathName.lastIndexOf("/") + 1)); // children里面的内容不带/，而path带/
    
            if(0 == index) { 
                // 获得锁
                countDownLatch.countDown();
            } else { 
                // 监听前一个节点的删除事件
                zooKeeper.exists(key + "/" + children.get(index - 1), this, this, "xxxx");
            }
    
        }
    
        // exists中监听事件，主要监听删除
        @Override
        public void process(WatchedEvent watchedEvent) { 
            switch (watchedEvent.getType()) { 
    
                case None:
                    break;
                case NodeCreated:
                    break;
                case NodeDeleted:
                    // 前面的节点被删除了，有可能是释放了锁，也有可能是客户端挂了
                    // countDownLatch.countDown(); // 这里不能直接获取锁，还得再去看看前面有没有人
                    zooKeeper.getChildren(key, false, this, "");
                    break;
                case NodeDataChanged:
                    break;
                case NodeChildrenChanged:
                    break;
            }
        }
    
        // exists的回调
        @Override
        public void processResult(int rc, String path, Object ctx, Stat stat) { 
            if(null == stat) {  // 前面的节点不存在了
                countDownLatch.countDown();
            }
        }
    
        public void unlock() { 
            try { 
                zooKeeper.delete(pathName, -1);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (KeeperException e) { 
                e.printStackTrace();
            }
        }
    }

### 简单使用 ###

package com.morris.distribute.lock.zookeeper.my;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.ZooDefs;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.data.Stat;
    import org.junit.Before;
    import org.junit.Test;
    
    import java.io.IOException;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.CyclicBarrier;
    import java.util.concurrent.TimeUnit;
    import java.util.stream.IntStream;
    
    @Slf4j
    public class ZookeeperLockDemo { 
    
        private ZooKeeper zooKeeper;
    
        @Before
        public void init() throws InterruptedException, IOException, KeeperException { 
            CountDownLatch countDownLatch = new CountDownLatch(1);
            ZooKeeper zooKeeper = new ZooKeeper("127.0.0.1:2181/zookeeperLock", 3_000, watchedEvent -> { 
    
                switch (watchedEvent.getState()) { 
                    case Unknown:
                        break;
                    case Disconnected:
                        break;
                    case NoSyncConnected:
                        break;
                    case SyncConnected:
                        countDownLatch.countDown();
                        break;
                    case AuthFailed:
                        break;
                    case ConnectedReadOnly:
                        break;
                    case SaslAuthenticated:
                        break;
                    case Expired:
                        break;
                }
            });
            countDownLatch.await();
            this.zooKeeper = zooKeeper;
    
            Stat exists = zooKeeper.exists("/", false);
    
            if(null == exists) { 
                // 创建所有锁的根节点，在连接字符串中
                zooKeeper.create("/", "xxx".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }
    
        @Test
        public void test() throws InterruptedException, KeeperException { 
            String root = "/updateStatus001";
    
            Stat exists = zooKeeper.exists(root, false);
    
            if(null == exists) { 
                // 创建每把锁的根节点
                zooKeeper.create(root, "xxx".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
    
            CyclicBarrier cyclicBarrier = new CyclicBarrier(10);
            IntStream.rangeClosed(1, 10).forEach((i) -> new Thread(() -> { 
    
                ZookeeperLock lock = new ZookeeperLock(zooKeeper);
    
                try { 
                    cyclicBarrier.await();
                    lock.lock(root); // 加锁
                } catch (Exception e) { 
                    e.printStackTrace();
                }
                try { 
                    log.info("get lock");
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    lock.unlock(); // 释放锁
    
                }
            }, "t" + i).start());
    
            TimeUnit.SECONDS.sleep(100); // 保证上面10个线程跑起来
        }
    }

由于ZooKeeper是异步建立连接，所以需要使用CountDownLatch来阻塞，等待连接获取成功后再返回。

另外这里的zookeeper的连接字符串中带有目录`/zookeeperLock`，这样create、get等命令中的路径都是相对于`/zookeeperLock`目录，而不是`/`。

运行结果如下：

2020-09-17 17:30:40,517  INFO [t1] (ZookeeperLockDemo.java:80) - get lock
    2020-09-17 17:30:45,553  INFO [t5] (ZookeeperLockDemo.java:80) - get lock
    2020-09-17 17:30:50,582  INFO [t9] (ZookeeperLockDemo.java:80) - get lock
    2020-09-17 17:30:55,623  INFO [t4] (ZookeeperLockDemo.java:80) - get lock
    2020-09-17 17:31:00,653  INFO [t3] (ZookeeperLockDemo.java:80) - get lock
    2020-09-17 17:31:05,683  INFO [t6] (ZookeeperLockDemo.java:80) - get lock
    2020-09-17 17:31:10,715  INFO [t2] (ZookeeperLockDemo.java:80) - get lock
    2020-09-17 17:31:15,756  INFO [t10] (ZookeeperLockDemo.java:80) - get lock
    2020-09-17 17:31:20,790  INFO [t8] (ZookeeperLockDemo.java:80) - get lock
    2020-09-17 17:31:25,831  INFO [t7] (ZookeeperLockDemo.java:80) - get lock

由于这里只是简单模拟，在生产应用中推荐使用curator提供的成熟工具类，所以这里并没有考虑锁的重入，锁的可中断，锁的超时等机制，有兴趣的可自行深入研究。