分布式锁—7.Curator的分布式锁

大纲
1.Curator的可重入锁的源码
2.Curator的非可重入锁的源码
3.Curator的可重入读写锁的源码
4.Curator的MultiLock源码
5.Curator的Semaphore源码
 
1.Curator的可重入锁的源码
(1)InterProcessMutex获取分布式锁
(2)InterProcessMutex的初始化
(3)InterProcessMutex.acquire()尝试获取锁
(4)LockInternals.attemptLock()尝试获取锁
(5)不同客户端线程获取锁时的互斥实现
(6)同一客户端线程可重入加锁的实现
(7)客户端线程释放锁的实现
(8)客户端线程释放锁后其他线程获取锁的实现
(9)InterProcessMutex就是一个公平锁
 
(1)InterProcessMutex获取分布式锁

1. public class Demo {
2.     public static void main(String[] args) throws Exception {
3.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
4.         CuratorFramework client = CuratorFrameworkFactory.newClient(
5.             "127.0.0.1:2181",
6.             5000,
7.             3000,
8.             retryPolicy
9.         );
10.         client.start();
11.         System.out.println("已经启动Curator客户端");
12.         
13.         //获取分布式锁
14.         InterProcessMutex lock = new InterProcessMutex(client, "/locks/myLock");
15.         lock.acquire();
16.         Thread.sleep(1000);
17.         lock.release();
18.     }
19. }

复制代码

(2)InterProcessMutex的初始化
设置锁的节点路径basePath + 初始化一个LockInternals对象实例。

1. public class InterProcessMutex implements InterProcessLock, Revocable<InterProcessMutex> {
2.     private final LockInternals internals;
3.     private final String basePath;
4.     private static final String LOCK_NAME = "lock-";
5.     ...
6.     public InterProcessMutex(CuratorFramework client, String path) {
7.         this(client, path, new StandardLockInternalsDriver());
8.     }
9.    
10.     public InterProcessMutex(CuratorFramework client, String path, LockInternalsDriver driver) {
11.         this(client, path, LOCK_NAME, 1, driver);
12.     }
13.    
14.     //初始化InterProcessMutex
15.     InterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver) {
16.         //1.设置锁的节点路径
17.         basePath = PathUtils.validatePath(path);
18.         //2.初始化一个LockInternals对象实例
19.         internals = new LockInternals(client, driver, path, lockName, maxLeases);
20.     }
21. }
23. public class LockInternals {
24.     private final LockInternalsDriver driver;
25.     private final String lockName;
26.     private volatile int maxLeases;
27.     private final WatcherRemoveCuratorFramework client;
28.     private final String basePath;
29.     private final String path;
30.     ...
31.     LockInternals(CuratorFramework client, LockInternalsDriver driver, String path, String lockName, int maxLeases) {
32.         this.driver = driver;
33.         this.lockName = lockName;
34.         this.maxLeases = maxLeases;
35.         this.client = client.newWatcherRemoveCuratorFramework();
36.         this.basePath = PathUtils.validatePath(path);
37.         this.path = ZKPaths.makePath(path, lockName);
38.     }
39.     ...
40. }

复制代码

(3)InterProcessMutex.acquire()尝试获取锁
LockData是InterProcessMutex的一个静态内部类。一个线程对应一个LockData实例对象，用来描述线程持有的锁的具体情况。多个线程对应的LockData存放在一个叫threadData的ConcurrentMap中。LockData中有一个原子变量lockCount，用于锁的重入次数计数。
 
在执行InterProcessMutex的acquire()方法尝试获取锁时：首先会尝试取出当前线程对应的LockData数据，判断是否存在。如果存在，则说明锁正在被当前线程重入，重入次数自增后直接返回。如果不存在，则调用LockInternals的attemptLock()方法尝试获取锁。默认情况下，attemptLock()方法传入的等待获取锁的时间time = -1。

1. public class InterProcessMutex implements InterProcessLock, Revocable<InterProcessMutex> {
2.     private final LockInternals internals;
3.     private final String basePath;
4.     private static final String LOCK_NAME = "lock-";
5.     //一个线程对应一个LockData数据对象
6.     private final ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap();
7.     ...
8.     //初始化InterProcessMutex
9.     InterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver) {
10.         //设置锁的路径
11.         basePath = PathUtils.validatePath(path);
12.         //初始化LockInternals
13.         internals = new LockInternals(client, driver, path, lockName, maxLeases);
14.     }
15.    
16.     @Override
17.     public void acquire() throws Exception {
18.         //获取分布式锁，会一直阻塞等待直到获取成功
19.         //相同的线程可以重入锁，每一次调用acquire()方法都要匹配一个release()方法的调用
20.         if (!internalLock(-1, null)) {
21.             throw new IOException("Lost connection while trying to acquire lock: " + basePath);
22.         }
23.     }
24.    
25.     private boolean internalLock(long time, TimeUnit unit) throws Exception {
26.         //获取当前线程
27.         Thread currentThread = Thread.currentThread();
28.         //获取当前线程对应的LockData数据
29.         LockData lockData = threadData.get(currentThread);
30.         if (lockData != null) {
31.             //可重入计算
32.             lockData.lockCount.incrementAndGet();
33.             return true;
34.         }
35.         //调用LockInternals.attemptLock()方法尝试获取锁，默认情况下，传入的time=-1，表示等待获取锁的时间
36.         String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
37.         if (lockPath != null) {
38.             //获取锁成功，将当前线程 + 其创建的临时顺序节点路径，封装成一个LockData对象
39.             LockData newLockData = new LockData(currentThread, lockPath);
40.             //然后把该LockData对象存放到InterProcessMutex.threadData这个Map中
41.             threadData.put(currentThread, newLockData);
42.             return true;
43.         }
44.         return false;
45.     }
46.    
47.     //LockData是InterProcessMutex的一个静态内部类
48.     private static class LockData {
49.         final Thread owningThread;
50.         final String lockPath;
51.         final AtomicInteger lockCount = new AtomicInteger(1);//用于锁的重入次数计数
52.         private LockData(Thread owningThread, String lockPath) {
53.             this.owningThread = owningThread;
54.             this.lockPath = lockPath;
55.         }
56.     }
57.    
58.     protected byte[] getLockNodeBytes() {
59.         return null;
60.     }
61.     ...
62. }

复制代码

(4)LockInternals.attemptLock()尝试获取锁
先创建临时节点，再判断是否满足获取锁的条件。
 
步骤一：首先调用LockInternalsDriver的createsTheLock()方法创建一个临时顺序节点。其中creatingParentContainersIfNeeded()表示级联创建，forPath(path)表示创建的节点路径名称，withMode(CreateMode.EPHEMERAL_SEQUENTIAL)表示临时顺序节点。
 
步骤二：然后调用LockInternals的internalLockLoop()方法检查是否获取到了锁。在LockInternals的internalLockLoop()方法的while循环中，会先获取排好序的客户端线程尝试获取锁时创建的临时顺序节点名称列表。然后获取当前客户端线程尝试获取锁时创建的临时顺序节点的名称，再根据名称获取在节点列表中的位置 + 是否可以获取锁 + 前一个节点的路径，也就是获取一个封装好这些信息的PredicateResults对象。
 
具体会根据节点名称获取当前线程创建的临时顺序节点在节点列表的位置，然后会比较当前线程创建的节点的位置和maxLeases的大小。其中maxLeases代表了同时允许多少个客户端可以获取到锁，默认是1。如果当前线程创建的节点的位置小，则表示可以获取锁。如果当前线程创建的节点的位置大，则表示获取锁失败。
 
获取锁成功，则会中断LockInternals的internalLockLoop()方法的while循环，然后向外返回当前客户端线程创建的临时顺序节点路径。接着在InterProcessMutex的internalLock()方法中，会将当前线程 + 其创建的临时顺序节点路径，封装成一个LockData对象，然后把该LockData对象存放到InterProcessMutex.threadData这个Map中。
 
获取锁失败，则通过PredicateResults对象先获取前一个节点路径名称。然后通过getData()方法获取前一个节点路径在zk的信息，并添加Watcher监听。该Watcher监听主要是用来唤醒在LockInternals中被wait()阻塞的线程。添加完Watcher监听后，便会调用wait()方法将当前线程挂起。
 
所以前一个节点发生变化时，便会通知添加的Watcher监听。然后便会唤醒阻塞的线程，继续执行internalLockLoop()方法的while循环。while循环又会继续获取排序的节点列表 + 判断当前线程是否已获取锁。

1. public class LockInternals {
2.     private final LockInternalsDriver driver;
3.     LockInternals(CuratorFramework client, LockInternalsDriver driver, String path, String lockName, int maxLeases) {
4.         this.driver = driver;
5.         this.path = ZKPaths.makePath(path, lockName);//生成要创建的临时节点路径名称
6.         ...
7.     }
8.     ...
9.     String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception {
10.         //获取当前时间
11.         final long startMillis = System.currentTimeMillis();
12.         //默认情况下millisToWait=null
13.         final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;
14.         //默认情况下localLockNodeBytes也是null
15.         final byte[] localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
16.         int retryCount = 0;
17.      
18.         String ourPath = null;
19.         boolean hasTheLock = false;//是否已经获取到锁
20.         boolean isDone = false;//是否正在获取锁
21.         while (!isDone) {
22.             isDone = true;
23.             //1.这里是关键性的加锁代码，会去级联创建一个临时顺序节点
24.             ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
25.             //2.检查是否获取到了锁
26.             hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
27.         }
28.         if (hasTheLock) {
29.             return ourPath;
30.         }
31.         return null;
32.     }
33.    
34.     private final Watcher watcher = new Watcher() {
35.         @Override
36.         public void process(WatchedEvent event) {
37.             //唤醒LockInternals中被wait()阻塞的线程
38.             client.postSafeNotify(LockInternals.this);
39.         }
40.     };
41.    
42.     //检查是否获取到了锁
43.     private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception {
44.         boolean haveTheLock = false;
45.         boolean doDelete = false;
46.         ...
47.         while ((client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock) {
48.             //3.获取排好序的各个客户端线程尝试获取分布式锁时创建的临时顺序节点名称列表
49.             List<String> children = getSortedChildren();
50.             //4.获取当前客户端线程尝试获取分布式锁时创建的临时顺序节点的名称
51.             String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
52.             //5.获取当前线程创建的节点在节点列表中的位置 + 是否可以获取锁 + 前一个节点的路径名称
53.             PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
54.             if (predicateResults.getsTheLock()) {//获取锁成功
55.                 //返回true
56.                 haveTheLock = true;
57.             } else {//获取锁失败
58.                 //获取前一个节点路径名称
59.                 String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
60.                 synchronized(this) {
61.                     //use getData() instead of exists() to avoid leaving unneeded watchers which is a type of resource leak
62.                     //通过getData()获取前一个节点路径在zk的信息，并添加watch监听
63.                     client.getData().usingWatcher(watcher).forPath(previousSequencePath);
64.                     //默认情况下，millisToWait = null
65.                     if (millisToWait != null) {
66.                         millisToWait -= (System.currentTimeMillis() - startMillis);
67.                         startMillis = System.currentTimeMillis();
68.                         if (millisToWait <= 0) {
69.                             doDelete = true;//timed out - delete our node
70.                             break;
71.                         }
72.                         wait(millisToWait);//阻塞
73.                     } else {
74.                         wait();//阻塞
75.                     }
76.                 }
77.             }
78.         }
79.         ...
80.         return haveTheLock;
81.     }
82.    
83.     List<String> getSortedChildren() throws Exception {
84.         //获取排好序的各个客户端线程尝试获取分布式锁时创建的临时顺序节点名称列表
85.         return getSortedChildren(client, basePath, lockName, driver);
86.     }
87.    
88.     public static List<String> getSortedChildren(CuratorFramework client, String basePath, final String lockName, final LockInternalsSorter sorter) throws Exception {
89.         //获取各个客户端线程尝试获取分布式锁时创建的临时顺序节点名称列表
90.         List<String> children = client.getChildren().forPath(basePath);
91.         //对节点名称进行排序
92.         List<String> sortedList = Lists.newArrayList(children);
93.         Collections.sort(
94.             sortedList,
95.             new Comparator<String>() {
96.                 @Override
97.                 public int compare(String lhs, String rhs) {
98.                     return sorter.fixForSorting(lhs, lockName).compareTo(sorter.fixForSorting(rhs, lockName));
99.                 }
100.             }
101.         );
102.         return sortedList;
103.     }
104.     ...
105. }
107. public class StandardLockInternalsDriver implements LockInternalsDriver {
108.     ...
109.     //级联创建一个临时顺序节点
110.     @Override
111.     public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception {
112.         String ourPath;
113.         //默认情况下传入的lockNodeBytes=null
114.         if (lockNodeBytes != null) {
115.             ourPath = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, lockNodeBytes);
116.         } else {
117.             //创建临时顺序节点
118.             ourPath = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
119.         }
120.         return ourPath;
121.     }
122.    
123.     //获取当前线程创建的节点在节点列表中的位置以及是否可以获取锁
124.     @Override
125.     public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception {
126.         //根据节点名称获取当前线程创建的临时顺序节点在节点列表中的位置
127.         int ourIndex = children.indexOf(sequenceNodeName);
128.         validateOurIndex(sequenceNodeName, ourIndex);
129.         //maxLeases代表的是同时允许多少个客户端可以获取到锁
130.         //getsTheLock为true表示可以获取锁，getsTheLock为false表示获取锁失败
131.         boolean getsTheLock = ourIndex < maxLeases;
132.         //获取当前节点需要watch的前一个节点路径
133.         String pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);
134.         return new PredicateResults(pathToWatch, getsTheLock);
135.     }
136.     ...
137. }

复制代码

(8)客户端线程释放锁后其他线程获取锁的实现
由于在节点列表里排第二的节点对应的线程会监听排第一的节点，而当持有锁的客户端线程释放锁后，排第一的节点会被删除掉。所以在节点列表里排第二的节点对应的客户端，便会收到zk的通知。于是会回调执行该线程添加的Watcher的process()方法，也就是唤醒该线程，让其继续执行while循环获取锁。
[code]public class LockInternals {    ...    private final Watcher watcher = new Watcher() {        @Override        public void process(WatchedEvent event) {            //唤醒LockInternals中被wait()阻塞的线程            client.postSafeNotify(LockInternals.this);        }    };        //检查是否获取到了锁    private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception {        boolean haveTheLock = false;        boolean doDelete = false;        ...        while ((client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock) {            //3.获取排好序的各个客户端线程尝试获取分布式锁时创建的临时顺序节点名称列表            List children = getSortedChildren();            //4.获取当前客户端线程尝试获取分布式锁时创建的临时顺序节点的名称            String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash            //5.获取当前线程创建的节点在节点列表中的位置+是否可以获取锁+前一个节点的路径名称            PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);            if (predicateResults.getsTheLock()) {//获取锁成功                //返回true                haveTheLock = true;            } else {//获取锁失败                //获取前一个节点路径名称                String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();                synchronized(this) {                    //use getData() instead of exists() to avoid leaving unneeded watchers which is a type of resource leak                    //通过getData()获取前一个节点路径在zk的信息，并添加watch监听                    client.getData().usingWatcher(watcher).forPath(previousSequencePath);                    //默认情况下，millisToWait = null                    if (millisToWait != null) {                        millisToWait -= (System.currentTimeMillis() - startMillis);                        startMillis = System.currentTimeMillis();                        if (millisToWait  0, "qty cannot be 0");        ImmutableList.Builder builder = ImmutableList.builder();        boolean success = false;        try {            while (qty-- > 0) {                int retryCount = 0;                long startMillis = System.currentTimeMillis();                boolean isDone = false;                while (!isDone) {                    switch (internalAcquire1Lease(builder, startMs, hasWait, waitMs)) {                        case CONTINUE: {                            isDone = true;                            break;                        }                        case RETURN_NULL: {                            return null;                        }                        case RETRY_DUE_TO_MISSING_NODE: {                            if (!client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper())) {                                throw new KeeperException.NoNodeException("Sequential path not found - possible session loss");                            }                            //try again                            break;                        }                    }                }            }            success = true;        } finally {            if (!success) {                returnAll(builder.build());            }        }        return builder.build();    }        private InternalAcquireResult internalAcquire1Lease(ImmutableList.Builder builder, long startMs, boolean hasWait, long waitMs) throws Exception {        if (client.getState() != CuratorFrameworkState.STARTED) {            return InternalAcquireResult.RETURN_NULL;        }        if (hasWait) {            long thisWaitMs = getThisWaitMs(startMs, waitMs);            if (!lock.acquire(thisWaitMs, TimeUnit.MILLISECONDS)) {                return InternalAcquireResult.RETURN_NULL;            }        } else {            //1.首先获取一个分布式锁            lock.acquire();        }        Lease lease = null;        boolean success = false;        try {            //2.尝试获取Semaphore的Lease：创建一个临时顺序节点            PathAndBytesable createBuilder = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL);            String path = (nodeData != null) ? createBuilder.forPath(ZKPaths.makePath(leasesPath, LEASE_BASE_NAME), nodeData) : createBuilder.forPath(ZKPaths.makePath(leasesPath, LEASE_BASE_NAME));            String nodeName = ZKPaths.getNodeFromPath(path);            lease = makeLease(path);            ...            try {                synchronized(this) {                    for(;;) {                        List children;                        //3.获取./lease目录下的所有临时顺序节点，并添加watcher监听                        children = client.getChildren().usingWatcher(watcher).forPath(leasesPath);                        ...                        //4.判断临时顺序节点的数量是否大于maxLeases                        //maxLeases表示最多允许多少个客户端线程获取Semaphore的Lease                        if (children.size()