【Zookeeper】应用场景：实现分布式锁

本是古典何须时尚 2022-11-03 15:29 157阅读 0赞

首先，我们需要明白一个问题，**分布式锁是什么？**

*  同一个JVM进程内，多个线程同步执行 ==> synchronized，Lock等锁  
    ![img][]
 *  分布式下，不同JVM进程访问同一资源 ==> Redis，Zookeeper等中间件  
    ![img][img 1]

> 所以，一个多核CPU也可以认为是一个小型分布式系统，多核CPU的并发编程的问题也可以认为是微观的分布式锁问题。

那如何通过 Zookeeper 实现分布式锁呢？

### 1.同级节点唯一性 ###

1）因为Zookeeper的**同级节点唯一性**，所以多个进程来向zookeeper创建一个相同名称节点，只会有一个成功。所以，可以通过在 zk 中创建同名 znode 来实现加锁（注：这里应该采用临时节点，因为临时节点会在会话关闭时消失，而进程拿锁也是，你都关了，还拿什么锁）

2）失败的进程全部通过**watcher机制**来监听zookeeper这个子节点的变化，一旦监听到该节点的删除事件，则再次触发所有进程去写锁

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

这种方式虽然简单可行，但是会产生**惊群效应**，即等锁释放后，所有等待进程都会被唤醒（zookeeper发送大量子节点变化信息给所有等待的客户端），但最终只有一个客户端获得锁。如果在集群够大的情况下，会对zookeeper服务器性能产生较大影响。

> PS：通过 Redis 实现分布式锁
> 
>  *  Redis 的 String 类型的 setnx(key, value) ，只有当 key 不存在时才能设置成功；所以同一时刻如果有多个请求过来，最终也只有一个进程会拿锁
>  *  当拿锁失败的进程，会被设置休眠指定时间 – 尽量避免惊群效应
>  *  拿锁进程操作完后，会删除当前 key
>  *  阻塞的进程，待醒来后再次尝试 setnx(key，value)

那有什么更好的办法吗？答：利用 Zookpeer 的有序节点实现分布式锁

### 2.临时有序节点 ###

所有客户端都在zookeeper的指定节点下创建一个临时有序节点，越早创建的顺序编号越小

*  因此可以将最小的节点设置为获得锁
 *  失败的节点只用监听比自己小的节点，当比自己小的节点删除以后，客户端就会收到watcher事件，然后再判断自己是否是最小的是否获得锁。这样就不会导致惊群效应，因为每个客户端只监听一个节点

**InterProcessMutex**

curator对于锁这块做了一些封装，curator提供了 InterProcessMutex 这样一个api。除了分布式锁之外， 还提供了leader选举、分布式队列等常用的功能。

*  InterProcessMutex：分布式可重入排它锁
 *  InterProcessSemaphoreMutex：分布式排它锁
 *  InterProcessReadWriteLock：分布式读写锁

public static void main(String[] args) {
        
            CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().
                    connectString("39.105.136.112:2181").
                    sessionTimeoutMs(5000).
                    retryPolicy(new ExponentialBackoffRetry(1000, 3)).
                    build();
            curatorFramework.start();
    		
         	// 可重入排它锁 ，传入指定父节点
            final InterProcessMutex lock = new InterProcessMutex(curatorFramework, "locks/");
    
            for (int i = 0; i < 10; i++) {
        
                new Thread(() -> {
        
                    System.out.println(Thread.currentThread().getName() + "-> 尝试获得锁");
                    try {
        
                        lock.acquire(); // 加锁
                        System.out.println(Thread.currentThread().getName() + "-> 成功获得锁");
                    } catch (Exception e) {
        
                        e.printStackTrace();
                    }
    
                    try {
        
                        lock.release(); // 释放锁
                        System.out.println(Thread.currentThread().getName() + "-> 释放锁成功");
                    } catch (Exception e) {
        
                        e.printStackTrace();
                    }
                }, "t" + i).start();
            }
        }

[img]: /images/20221023/de28b0ac2edd4715bc7afe35740ff4dc.png
[img 1]: /images/20221023/89b5a9c2f75544d38f5ea4a34d982ca2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221023/ab41b401e92e4b2895c4328fa38d734b.png