ZK实现分布式锁的记录

一时失言乱红尘 2021-12-12 16:57 238阅读 0赞

实现分布式锁有很多种方式：

1：数据库方式： 谁先插入数据库值成功， 谁就获取锁。

2： 缓存方式

3: zookeeper 实现分布式锁， 推荐这种方式

zk实现分布式锁的原理及流程： 临时节点 + 事件通知 + 信号量， 临时节点的生命周期是连接断开则会删除。

zk有四种节点类型

1： 持久化节点

2：临时节点

3：持久化顺序节点

4： 临时顺序节点。

zk 的事件通知 watcher.

实现分布式锁 流程：

1： 连接zk

2: 创建临时节点， 如果创建成功， 则获取锁， 代码执行完毕， 释放锁(也就是关闭zk连接)。

如果创建失败，则代表有服务创建了节点获取了锁， 此时则使用信号量等待。

3： 节点删除的通知， 如果有节点被删除的通知过来了， 则唤醒信号量， 重新尝试获取锁。

代码：

1: 锁接口

/** 
     * ClassName: Lock
     * Function:  定义锁
     * date: 2019年7月10日 下午9:44:32
     * 
     * @author tangjiandong
     */
    public interface Lock {
    	
    	void  getLock();
    	
    	void  unlock();
    
    }

2： 锁的两个实现：

/** 
     * Project Name:distributed 
     * File Name:ZookeeperAbstractLock.java 
     * Package Name:com.tang.service 
     * Date:2019年7月10日 下午9:52:06 
     * Copyright (c) 2019,   All Rights Reserved. 
     * 
     */
    package com.tang.service;
    
    import java.util.concurrent.CountDownLatch;
    
    import org.I0Itec.zkclient.ZkClient;
    
    /** 
     * ClassName: ZookeeperAbstractLock
     * Function: TODO ADD FUNCTION. 
     * date: 2019年7月10日 下午9:52:06
     * 
     * @author tangjiandong
     */
    public abstract  class ZookeeperAbstractLock implements Lock {
    
    	private static final String  connectAddress = "127.0.0.1:2181"; //连接地址
    	private static final int timeOut = 5000;   //超时时间
    	protected static final String path = "/lock"; //创建的节点名称
    	protected ZkClient zkClient = new ZkClient(connectAddress); //连接建立
    	protected CountDownLatch countDownLatch = null; //信号量计数器
    	
    	
    	/**
    	 * 
    	 * @see com.tang.service.Lock#getLock()
    	 */
    	@Override
    	public void getLock()
    	{
    		   //是否获取锁
    		   if(trylock())
    		   {
    			   System.out.println("---------获取锁!");
    		   }else {
    			   //如果等待到了锁，则继续获取锁， 否则继续等待
    			   waitLock();
    			   //尝试获取锁
    			   getLock();
    		   }
    	}
        //是否获取锁
    	abstract boolean trylock();
    	
    	//等待锁
    	abstract void waitLock();
    	/**
    	 *关闭连接， 释放锁
    	 * @see com.tang.service.Lock#unlock()
    	 */
    	@Override
    	public void unlock() 
    	{
    		 if(null != zkClient)
    		 {
    			 zkClient.close();
    			 System.out.println("---------释放锁!");
    		 }
    	}
    
    }

/** 
     * Project Name:distributed 
     * File Name:ZookeeperDistributedLock.java 
     * Package Name:com.tang.service 
     * Date:2019年7月10日 下午10:10:23 
     * Copyright (c) 2019, All Rights Reserved. 
     * 
     */
    package com.tang.service;
    
    import java.util.concurrent.CountDownLatch;
    
    import org.I0Itec.zkclient.IZkDataListener;
    
    /** 
     * ClassName: ZookeeperDistributedLock
     * Function:   获取锁与等待锁的实现
     * date: 2019年7月10日 下午10:10:23
     * 
     * @author tangjiandong
     */
    public class ZookeeperDistributedLock extends ZookeeperAbstractLock{
    
    	/**
    	 * 获取锁
    	 * @return
    	 * @see com.tang.service.ZookeeperAbstractLock#trylock()
    	 */
    	@Override
    	boolean trylock() 
    	{
    		try {
    			zkClient.createEphemeral(path); //创建临时节点
    			return true; //创建成功则获取锁
    		} catch (Exception e) {
    			return false;  //创建失败则没有获取到锁， 需要等待
    		}
    	}
    
    	/**
    	 * 等待锁
    	 * @see com.tang.service.ZookeeperAbstractLock#waitLock()
    	 */
    	@Override
    	void waitLock()
    	{
    		//创建节点的监听
    		IZkDataListener iZkDataListener = new IZkDataListener() {
    			
    			//节点被删除的通知
    			@Override
    			public void handleDataDeleted(String dataPath) throws Exception {
    				if(null != countDownLatch) 
    				{
    					countDownLatch.countDown(); //计数器减一，唤醒等待
    				}
    			}
    			
    			//节点改变的通知
    			@Override
    			public void handleDataChange(String dataPath, Object data) throws Exception {
    				// TODO Auto-generated method stub
    				
    			}
    		};
    		
    		
    		//注册监听
    		try {
    			zkClient.subscribeDataChanges(path, iZkDataListener);
    		} catch (Exception e) {
    			System.out.println("---------节点注册监听异常!");
    		}
    
    		
    		
    		if(zkClient.exists(path))  //如果节点存在了，则等待，
    		{
    			countDownLatch = new CountDownLatch(1);
    			
    			try {
    				countDownLatch.await(); //可加上超时等待， 防止延迟通知导致死锁
                    //countDownLatch.await(1000, TimeUnit.SECONDS);
    			} catch (InterruptedException e) {
    				//节点等待失败
    				System.out.println("---------节点等待异常!");
    				e.printStackTrace();
    			}
    		}
    		
    		//删除监听
    		zkClient.unsubscribeDataChanges(path, iZkDataListener);
    	}
    
    }

3:调用

/** 
     * Project Name:distributed 
     * File Name:CreateNumber.java 
     * Package Name:com.tang.number 
     * Date:2019��7��9�� ����8:38:45 
     * Copyright (c) 2019, ���쳤����Ϸֹ�˾  All Rights Reserved. 
     * 
     */
    package com.tang.number;
    
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import com.tang.service.ZookeeperDistributedLock;
    
    /** 
     * ClassName: CreateNumber
     * Function: TODO ADD FUNCTION. 
     * date: 2019��7��9�� ����8:38:45
     * 
     * @author tangjiandong
     */
    public class CreateNumber implements Runnable{
    
    	private com.tang.service.Lock lock = new ZookeeperDistributedLock(); //创建锁
    	
    	/**
    	 * 
    	 * @see java.lang.Runnable#run()
    	 */
    	@Override
    	public void run() {
    		//synchronized (this) {
    		lock.getLock();
    		try {
    			getNumber();
    		} finally {
    			lock.unlock();
    		}
    			
    		//}
    	}
    	
    	public void  getNumber()
    	{
    		String createJjdbh = JjdbhUtils.createJjdbh(null);
    		System.out.println(createJjdbh);
    	}
    	
    	public static void main(String[] args) {
    		for (int i = 0; i < 100; i++) 
    		{
    			new Thread(  new CreateNumber()).start();
    		}
    		
    	}
    	
    
    }