java使用数据库和zookeeper实现分布式锁

本是古典何须时尚 2022-02-24 02:58 132阅读 0赞

做为java程序员，锁这个东西相信大家都不陌生。在程序运行中，当多个线程或者多个进程同时对同一条数据进行操作时。如果不对该条数据进行锁定，那么实际执行的结果很容易就会出现错误，和预期的结果不相符。

比如当100个线程同时对java中的一个域变量进行叠加操作。了解过jvm的都知道，线程对共享变量的操作，并不是将共享变量本身直接拿过来进行操作，而是取出变量的副本，对这个副本进行操作，完成后再将副本保存到变量主内存中。由此可想而知，如果在取出变量副本后，在没有保存到主内存前这一段时间如果有其他线程对该变量做了修改，那么变量的值肯定和预期的不一致了。

解决这个问题很简单： java本身就提供了很多基于jvm的锁机制，synchronized,lock,各种并发安全的容器以及类。我们只需要将该方法体锁住，或者更好一点针对数据加锁，这样可以同条数据同步，不同数据异步。

但是，这种实现方式无不依赖一样东西：那就是同一个jvm。 如果jvm不同，或者存在多个jvm那就行不通了。

很明显，绝大多数企业，互联网项目已经告别一个war包打天下的古老阶段了，分布式，集群已经是主流。那么如果在多个jvm的情况下保证数据的同步操作呢。其实现在针对分布式同步，已经有非常多的解决方案了，比如使用数据库实现分布式锁，redis等缓存实现，zookeeper节点实现等。

这里我们就简单说两种，一种是基于数据库实现的分布式锁，一种是zookeeper实现。

** 1、数据库实现分布式锁。**

思路很简单：建一张表，简单一点只要一个字段，在我们需要锁住某个接口方法或者某条数据时，只需要在这张表中添加一条数据，释放锁时，只需要删除这条数据就可以了。

如下：

@Service
    public  class MysqlLock extends AbstractLock {
    	
    	@Resource
    	private LockMapper mapper;
    	
    	//所有的线程都往数据库插入主键值相同的数据
    	private static final int LOCK_ID = 1;
    	
    	public  void  getLock(){
    		//尝试获得锁资源
    		if (tryLock()) {
    			System.out.println("##获取lock锁的资源####");
    		} else {
    			// 等待
    			waitLock();
    			// 重新获取锁资源
    			getLock();
    		}
    	}
    
    
    	//非阻塞式加锁
    	private boolean tryLock() {
    		try {
    			mapper.insert(LOCK_ID);
    		} catch (Exception e) {
    			return false;
    		}
    		return true;
    	}
    
    	//让当前线程休眠一段时间
    
    	private void waitLock() {
    		try {
    			Thread.currentThread().sleep(10);
    		} catch (InterruptedException e) {
    
    			e.printStackTrace();
    		}
    	}
    
    
    
    	public void unLock() {
    		mapper.deleteByPrimaryKey(LOCK_ID);
    	}
    }

使用的时候我们只需要这样：

public int  updateStudent(Student student){
    	getLock();
    	try{
    		return mapper.updateStudent(student);
    	}finally {
    		unLock();
    	}
    }

**2、   第二种我们通过zookeeper来实现分布式锁**

public class ZookeeperDistrbuteLock extends ZookeeperAbstractLock {
        private CountDownLatch countDownLatch = null;
    
        @Override
        //尝试获得锁
        public  boolean tryLock() {
            try {
                zkClient.createEphemeral(PATH);
                return true;
            } catch (Exception e) {
                return false;
            }
    
        }
    
        @Override
        public void waitLock() {
            IZkDataListener izkDataListener = new IZkDataListener() {
                //节点被删除了，则执行到下一步，发令枪打响
                public void handleDataDeleted(String path) throws Exception {
                    // 唤醒被等待的线程
                    if (countDownLatch != null) {
                        countDownLatch.countDown();
                    }
                }
                public void handleDataChange(String path, Object data) throws Exception {
    
                }
            };
            // 注册事件
            zkClient.subscribeDataChanges(PATH, izkDataListener);
    
            //如果节点存
            if (zkClient.exists(PATH)) {
                countDownLatch = new CountDownLatch(1);
                try {
                    //等待，一直等到接受到事件通知
                    countDownLatch.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 删除监听
            zkClient.unsubscribeDataChanges(PATH, izkDataListener);
        }
    
        public void unLock() {
            //释放锁
            if (zkClient != null) {
                zkClient.delete(PATH);
                zkClient.close();
                System.out.println("释放锁资源...");
            }
        }
    }

zookeeper实现分布式锁和数据库原理类似，不过zookeeper是基于节点监听机制来实现，上面这种zookeeper实现分布式锁很容易出现羊群效应，所谓的羊群效应就是在一瞬间有过多的请求去争抢一把锁。

下面是使用zookeeper的有序节点实现的分布式锁

public class ZookeeperDistrbuteLock2 extends ZookeeperAbstractLock {
        private CountDownLatch countDownLatch= null;
    
        private String beforePath;//当前请求的节点前一个节点
        private String currentPath;//当前请求的节点
    
        public ZookeeperDistrbuteLock2() {
            if (!this.zkClient.exists(PATH2)) {
                this.zkClient.createPersistent(PATH2);
            }
        }
    
    
        @Override
        public boolean  tryLock() {
            //如果currentPath为空则为第一次尝试加锁，第一次加锁赋值currentPath
            if(currentPath == null || currentPath.length()<= 0){
                //创建一个临时顺序节点
                currentPath = this.zkClient.createEphemeralSequential(PATH2 + '/',"lock");
            }
            //获取所有临时节点并排序，临时节点名称为自增长的字符串如：0000000400
            List<String> childrens = this.zkClient.getChildren(PATH2);
            Collections.sort(childrens);
    
            if (currentPath.equals(PATH2 + '/'+childrens.get(0))) {//如果当前节点在所有节点中排名第一则获取锁成功
                return true;
            } else {//如果当前节点在所有节点中排名中不是排名第一，则获取前面的节点名称，并赋值给beforePath
                int wz = Collections.binarySearch(childrens,
                        currentPath.substring(7));
                beforePath = PATH2 + '/'+childrens.get(wz-1);
            }
            return false;
    
        }
    
    
        @Override
        public void waitLock() {
            IZkDataListener listener = new IZkDataListener() {
    
                public void handleDataDeleted(String dataPath) throws Exception {
    
                    if(countDownLatch!=null){
                        countDownLatch.countDown();
                    }
                }
    
                public void handleDataChange(String dataPath, Object data) throws Exception {
    
                }
            };
            //给排在前面的的节点增加数据删除的watcher,本质是启动另外一个线程去监听前置节点
            this.zkClient.subscribeDataChanges(beforePath, listener);
    
            if(this.zkClient.exists(beforePath)){
                countDownLatch=new CountDownLatch(1);
                try {
                    countDownLatch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //删除监听
            this.zkClient.unsubscribeDataChanges(beforePath, listener);
        }
    
    
        public void unLock() {
            //删除当前临时节点
            zkClient.delete(currentPath);
            zkClient.close();
        }
    }

以上