Zookeeper-ZooKeeper 分布式锁案例

我不是女神ヾ 2022-09-08 13:00 162阅读 0赞

# ZooKeeper 分布式锁案例 #

什么叫做分布式锁呢?

比如说"进程 1"在使用该资源的时候,会先去获得锁,"进程 1"获得锁以后会对该资源保持独占,这样其他进程就无法访问该资源,"进程 1"用完该资源以后就将锁释放掉,让其他进程来获得锁,那么通过这个锁机制,我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。

那么我们把这个分布式环境下的这个锁叫作分布式锁

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

锁代码

public class DistributeLock { 
    
        // 注意逗号左右不能有空格
        private final String connectString = "hadoop113:2181,hadoop114:2181,hadoop115:2181";
        private int sessionTimeout = 2000;
        private ZooKeeper zkClient;
    
        private String currentMode;
        // 前一个节点
        private String waitPath;
    
        private CountDownLatch connectDownLatch = new CountDownLatch(1);
        private CountDownLatch waitDownLatch = new CountDownLatch(1);
    
        public DistributeLock() throws IOException, KeeperException, InterruptedException { 
    
            // 1、获取zk连接
            zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() { 
                public void process(WatchedEvent watchedEvent) { 
                    // connectDownLatch 如果连接上zk，可以释放
                    if (watchedEvent.getState() == Event.KeeperState.SyncConnected) { 
                        connectDownLatch.countDown();
                    }
    
                    // waitDownLatch 需要释放
                    if (watchedEvent.getType() == Event.EventType.NodeDeleted
                            && watchedEvent.getPath().equals(waitPath)) { 
                        waitDownLatch.countDown();
                    }
                }
            });
    
            // 等待zk正常连接后，程序再往下走
            connectDownLatch.await();
    
    
            // 2、判断根节点是否存在
            Stat stat = zkClient.exists("/locks", false);
            if (stat == null) { 
                // 创建一下节点，权限是所有人均可读的永久节点
                zkClient.create("/locks", "locks".getBytes(),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }
    
        // 对zk加锁
        public void zkLock() throws KeeperException, InterruptedException { 
    
            // 判断对应的临时带序号的临时节点
            currentMode = zkClient.create("/locks/" + "seq-", null,
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    
            // 判断创建的几点是否是最小的需要节点，如果是获取到锁；如果不是，监听它序号前一个节点
            List<String> children = zkClient.getChildren("/locks", false);
    
            // 如果children只有一个值，那就直接获取锁，如果有多个节点，需要判断谁最小
            if (children.size() == 1) { 
                return;
            } else { 
                // 排序
                Collections.sort(children);
    
                // 获取节点名称, seq-00000000
                String thisNode = currentMode.substring("/locks/".length());
    
                // 通过seq-seq-00000000获取到该节点在children集合的位置
                int index = children.indexOf(thisNode);
                if (index == -1) { 
                    System.out.println("数据异常");
                } else if (index == 0) { 
                    // 是第一个节点，可以获取锁了
                    return;
                } else { 
                    // 需要监听前一个节点变化
                    waitPath = "/locks/" + children.get(index - 1);
                    zkClient.getData(waitPath, true, null);
    
                    // 等待监听
                    waitDownLatch.await();
                }
            }
        }
    
        // 对zk解锁
        public void unZkLock() throws KeeperException, InterruptedException { 
    
            // 删除节点
            zkClient.delete(currentMode, -1);
        }
    }

测试代码

public class DistributedLockTest { 
    
        public static void main(String[] args) throws InterruptedException, IOException, KeeperException { 
    
            final DistributeLock lock1 = new DistributeLock();
            final DistributeLock lock2 = new DistributeLock();
    
            new Thread(new Runnable() { 
                public void run() { 
    
                    try { 
                        lock1.zkLock();
                        System.out.println("线程1启动，获得锁");
                        Thread.sleep(5 * 1000);
    
                        lock1.unZkLock();
                        System.out.println("线程1释放锁");
    
                    } catch (KeeperException e) { 
                        e.printStackTrace();
                    } catch (InterruptedException interruptedException) { 
                        interruptedException.printStackTrace();
                    }
                }
            }).start();
    
            new Thread(new Runnable() { 
                public void run() { 
    
                    try { 
                        lock2.zkLock();
                        System.out.println("线程2启动，获得锁");
                        Thread.sleep(5 * 1000);
    
                        lock2.unZkLock();
                        System.out.println("线程2释放锁");
    
                    } catch (KeeperException e) { 
                        e.printStackTrace();
                    } catch (InterruptedException interruptedException) { 
                        interruptedException.printStackTrace();
                    }
                }
            }).start();
    
            // 线程2启动，获得锁
            // 线程2释放锁
            // 线程1启动，获得锁
            // 线程1释放锁
        }
    }

## Curator 框架实现分布式锁案例 ##

1)原生的 Java API 开发存在的问题

(1)会话连接是异步的,需要自己去处理。比如使用 CountDownLatch

(2)Watch 需要重复注册,不然就不能生效

(3)开发的复杂性还是比较高的

(4)不支持多节点删除和创建。需要自己去递归

2)Curator 是一个专门解决分布式锁的框架,解决了原生 Java API 开发分布式遇到的问题。

详情请查看官方文档:https://curator.apache.org/index.html

3)Curator 案例实操

添加依赖

<dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>4.3.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>4.3.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-client</artifactId>
        <version>4.3.0</version>
    </dependency>

public class CuratorLockTest { 
    
        public static void main(String[] args) { 
    
            // 创建分布式锁1
            final InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");
    
            // 创建分布式锁2
            final InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");
    
            new Thread(new Runnable() { 
                public void run() { 
    
                    try { 
                        lock1.acquire();
                        System.out.println("线程1获得锁");
    
                        lock1.acquire();
                        System.out.println("线程1再次获得锁");
                        Thread.sleep(5 * 1000);
    
                        lock1.release();
                        System.out.println("线程1释放锁");
    
                        lock1.release();
                        System.out.println("线程1再次释放锁");
    
                    } catch (KeeperException e) { 
                        e.printStackTrace();
                    } catch (InterruptedException interruptedException) { 
                        interruptedException.printStackTrace();
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            }).start();
    
            new Thread(new Runnable() { 
                public void run() { 
    
                    try { 
                        lock2.acquire();
                        System.out.println("线程2获得锁");
    
                        lock2.acquire();
                        System.out.println("线程2再次获得锁");
                        Thread.sleep(5 * 1000);
    
                        lock2.release();
                        System.out.println("线程2释放锁");
    
                        lock2.release();
                        System.out.println("线程2再次释放锁");
    
                    } catch (KeeperException e) { 
                        e.printStackTrace();
                    } catch (InterruptedException interruptedException) { 
                        interruptedException.printStackTrace();
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            }).start();
    
            // 线程1获得锁
            // 线程1再次获得锁
            // 线程1释放锁
            // 线程1再次释放锁
            // 线程2获得锁
            // 线程2再次获得锁
            // 线程2释放锁
            // 线程2再次释放锁
    
        }
    
        private static CuratorFramework getCuratorFramework() { 
    
            ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
    
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .connectString("hadoop113:2181,hadoop114:2181,hadoop115:2181")
                    .connectionTimeoutMs(2000)
                    .sessionTimeoutMs(2000)
                    .retryPolicy(policy)
                    .build();
    
            client.start();
    
            System.out.println("zookeeper连接启动成功");
    
            return client;
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBA6a2U56ybTG92ZQ_size_23_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/5d89330069ec4213a70b3558c7c32d99.png