Hadoop-32 ZooKeeper 分布式锁问题 分布式锁Java实现 附带案例和实现思路代码

avatar
作者
筋斗云
阅读量:2

章节内容

上节我们完成了:

  • ZooKeeper的Leader选举机制
  • ZooKeeper的选举过程
  • ZooKeeper的ZAB协议

背景介绍

这里是三台公网云服务器,每台 2C4G,搭建一个Hadoop的学习环境,供我学习。
之前已经在 VM 虚拟机上搭建过一次,但是没留下笔记,这次趁着前几天薅羊毛的3台机器,赶紧尝试在公网上搭建体验一下。

  • 2C4G 编号 h121
  • 2C4G 编号 h122
  • 2C2G 编号 h123

分布式锁

出现问题1(单机器)

  • 假设 Redis 里面的某个商品库存为1,此时两个用户同时下单,其中一个下单请求执行到第3步,更新数据库的库存为0,但是第4步还没执行。
  • 而另外一个用户下单执行到了第二步,发现库存还是1,就会继续执行第3步。
  • 但是此时库存已经为0了,所以数据库没有限制,此时会出现超卖的问题。
    在这里插入图片描述

解决方案1

  • 用锁把2、3、4步锁住,让他们执行完后,另一个线程才能够继续执行。
  • 但是由于业务发展迅速,原来的单机已经不能够满足,此时增加一台机器后,会出现更严重的问题。
    在这里插入图片描述

出现问题2(多机器)

假设有两个订单同时执行,分别有两个机器执行,那么这两个请求就是可以同时执行了,这样就依然出现了超卖的问题。

在这里插入图片描述

解决方案2

我们需要使用分布式锁来解决上面出现的问题。
分布式锁的作用就是在整个系统中提供一个全局的、唯一的锁,在分布式系统中每个系统进行相关的操作时都需要获取到该锁,才能够执行相应的操作。

ZK 分布式锁

实现思路

  • 锁就是ZK指定目录下序号最小的临时节点,多个系统的多个线程都要在此目录下创建临时顺序节点,因为ZK会保证节点的顺序性,所以可以利用节点的顺序性进行锁判断。
  • 每个线程都是先创建临时顺序节点,然后获取当前目录下最小的节点(序号),判断最小节点是不是当前节点,如果是那么获取锁成功,如果不是则获取锁失败。
  • 获取锁失败的线程获取当前节点上一个临时顺序节点,并对此节点进行监听,当该节点删除时,代表释放了锁。

流程图

在这里插入图片描述

编写代码

LockTest

package icu.wzk.zk.demo02;  public class LockTest {      public static void main(String[] args) {         for (int i = 0; i < 10; i ++) {             // 启动10个             new Thread(new LockRunnable()).start();         }     }      static class LockRunnable implements Runnable {          @Override         public void run() {             final ClientTest clientTest = new ClientTest();             clientTest.getLock();             try {                 Thread.sleep(2000);             } catch (InterruptedException e) {                 e.printStackTrace();             }             clientTest.deleteLock();         }     }  }  

ClientTest

package icu.wzk.zk.demo02;  import org.I0Itec.zkclient.IZkDataListener; import org.I0Itec.zkclient.ZkClient;  import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch;  public class ClientTest {      private ZkClient zkClient = new ZkClient("h121.wzk.icu:2181,h122.wzk.icu:2181,h123.wzk.icu:2181");      String beforeNodePath;      String currentNodePath;      CountDownLatch countDownLatch = null;      public ClientTest() {         synchronized (ClientTest.class) {             if (!zkClient.exists("/lock")) {                 zkClient.createPersistent("/lock");             }         }     }      public boolean tryGetLock() {         if (null == currentNodePath || currentNodePath.isEmpty()) {             currentNodePath = zkClient.createEphemeralSequential("/lock/", "lock");         }         final List<String> childs = zkClient.getChildren("/lock");         Collections.sort(childs);         final String minNode = childs.get(0);         if (currentNodePath.equals("/lock/" + minNode)) {             return true;         } else {             final int i = Collections.binarySearch(childs, currentNodePath.substring("/lock/".length()));             String lastNodeChild = childs.get(i - 1);             beforeNodePath = "/lock/" + lastNodeChild;         }         return false;     }      public void waitForLock() {         final IZkDataListener iZkDataListener = new IZkDataListener() {             @Override             public void handleDataChange(String dataPath, Object data) throws Exception {                 //             }              @Override             public void handleDataDeleted(String dataPath) throws Exception {                 countDownLatch.countDown();             }         };         zkClient.subscribeDataChanges(beforeNodePath, iZkDataListener);          if (zkClient.exists(beforeNodePath)) {             countDownLatch = new CountDownLatch(1);             try {                 countDownLatch.await();             } catch (InterruptedException e) {                 e.printStackTrace();             }         }          zkClient.unsubscribeDataChanges(beforeNodePath, iZkDataListener);     }      public void deleteLock() {         if (zkClient != null) {             zkClient.delete(currentNodePath);             zkClient.close();         }     }      public void getLock() {         final String threadName = Thread.currentThread().getName();         if (tryGetLock()) {             System.out.println(threadName + ": 获取到了锁!");         } else {             System.out.println(threadName + ": 没有获取到锁!");             waitForLock();             // 自己调用自己             getLock();         }     }   }  

运行结果

在这里插入图片描述

广告一刻

为您即时展示最新活动产品广告消息,让您随时掌握产品活动新动态!