分布式锁的应用

- 日理万妓 2022-06-06 07:56 191阅读 0赞

之前说到redis队列的场景使用lua脚本可以解决高并发的原子性问题，但是redis集群并不支持lua脚本，因此想要实现原子性防止高并发的问题就需要使用分布式锁。

这里总结两种分布式锁的实现，使用Redisson和Zookeeper。

一、场景回顾

先回顾一下场景，我们假设两个线程分别为不同jvm上的两个程序，一个服务在入队，一个服务在出队，看看判断入队的逻辑是否可靠

public static void main(String[] args) {
    		
    		CountDownLatch countDownLatch = new CountDownLatch(1);
    		
    		LockTest lockTest = new LockTest();
    		
    		Runnable inQueue = ()->{
    			for(int i=0; i<100; i++) {
    				try {
    					lockTest.inQueue(i);
    					countDownLatch.countDown();
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    		};
    		
    		Runnable outQueue = ()->{
    			try {
    				countDownLatch.await();
    				for(int i=0; i<30; i++) {
    					lockTest.outQueue();
    				}
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		};
    		
    		new Thread(inQueue).start();
    		new Thread(outQueue).start();
    		
    	}

![Center][]

我们发现，很可能会出现判断出错的情况

二、Redisson实现分布式锁

我们使用Redisson加锁，可以解决这个问题

Redisson工具类

public class RedissonUtil {
    
    	private static Config config = new Config();
        private static Redisson redisson = null;
    
        // 不要在分布式环境里用静态块初始化配置
        public static void init(){
            try {
    //            config.useClusterServers() //这是用的集群server
    //                    .setScanInterval(2000) //设置集群状态扫描时间
    //                    .setMasterConnectionPoolSize(200) //设置连接数
    //                    .setSlaveConnectionPoolSize(300)
    //                    dev
    //                    .addNodeAddress("redis://@127.0.0.1:6379");
    //                    product
    //                    .addNodeAddress("redis://@120.76.194.117:6379");
                
                config.useSingleServer()
                	  .setConnectionPoolSize(200)
                	  .setAddress("redis://@127.0.0.1:6379")
                	  .setPassword("123456");
                	  
                redisson = (Redisson) Redisson.create(config);
                System.out.println(redisson);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        
        public static Redisson getRedisson(){
            return redisson;
        }
        
        public static <V> RDeque<V> getRDeque(String key){  
            RDeque<V> rDeque=redisson.getDeque(key);  
            return rDeque;  
        }  
        
    }

分布式锁工具类

public class DistributedRedisLock {
    
    	private static Logger logger = Logger.getLogger(DistributedRedisLock.class.getName());
    	
    	private static final String LOCK_TITLE = "lock_";
    	
    	public static void acquire(String lockName){
    		Redisson redisson = RedissonUtil.getRedisson();
    		logger.log(Level.INFO, "加分布式锁:" + LOCK_TITLE + lockName);
                    String key = LOCK_TITLE + lockName;
                    RLock mylock = redisson.getLock(key);
                    mylock.lock(2, TimeUnit.MINUTES); 
                    logger.log(Level.INFO, "加分布式锁成功:" + LOCK_TITLE + lockName);
           }
    	
    	public static boolean acquireFaileToBreak(String lockName){
    		Redisson redisson = RedissonUtil.getRedisson();
    		logger.log(Level.INFO, "加分布式锁:" + LOCK_TITLE + lockName);
                    String key = LOCK_TITLE + lockName;
                    RLock mylock = redisson.getLock(key);
                    if(mylock.isLocked()) {
            	   return false;
                    }
                    mylock.lock(2, TimeUnit.MINUTES); 
                    logger.log(Level.INFO, "加分布式锁成功:" + LOCK_TITLE + lockName);
                    return true;
            }
    	
    	public static void release(String lockName){
    		Redisson redisson = RedissonUtil.getRedisson();
    		logger.log(Level.INFO, "释放分布式锁:" + LOCK_TITLE + lockName);
                    String key = LOCK_TITLE + lockName;
                    RLock mylock = redisson.getLock(key);
                    mylock.unlock();
                    logger.log(Level.INFO, "释放分布式锁成功:" + LOCK_TITLE + lockName);
           }
    	
    }

public static void main(String[] args) {
    		
    		CountDownLatch countDownLatch = new CountDownLatch(1);
    		
    		RedissonUtil.init();
    		
    		JedisUtil jedisUtil = new JedisUtil();
    		LockTest lockTest = new LockTest();
    		
    		Runnable inQueue = ()->{
    			for(int i=0; i<100; i++) {
    				try {
    					lockTest.inQueue(i);
    					countDownLatch.countDown();
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    		};
    		
    		Runnable outQueue = ()->{
    			try {
    				countDownLatch.await();
    				for(int i=0; i<30; i++) {
    					lockTest.outQueue();
    				}
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		};
    		
    		new Thread(inQueue).start();
    		new Thread(outQueue).start();
    		
    	}

public class LockTest {
    	private ReentrantLock lock = new ReentrantLock();
    	
    	public void inQueueWithScript(int value) {
    		System.out.println(Thread.currentThread().getName()+" inQueue");
    		long success = (long) JedisUtil.evalScript(LuaSysContants.IN_QUEUE.getName(), LuaSysContants.IN_QUEUE.getScript(), 1, "result_list", String.valueOf(value));
    		if(success == 0) {
    			System.err.println("入队出错！");
    		}
    	}
    	
    	public void inQueue(int value) {
    		// 分布式Redisson锁
    		DistributedRedisLock.acquire("lock");
    		try {
    			System.out.println(Thread.currentThread().getName()+" rpush");
    			long retValue = JedisUtil.rpush("result_list", String.valueOf(value));
    			System.out.println(Thread.currentThread().getName()+" llen");
    			long len = JedisUtil.llen("result_list");
    			if(retValue != len) {
    				System.err.println(retValue+":"+len);
    				System.err.println("入队出错！");
    			}
    		} finally {
    			DistributedRedisLock.release("lock");
    		}
    	}
    	
    	public void outQueue() {
    		// 分布式Redisson锁
    		DistributedRedisLock.acquire("lock");
    		try {
    			System.out.println(Thread.currentThread().getName()+" lpop");
    			JedisUtil.lpop("result_list");
    		} finally {
    			DistributedRedisLock.release("lock");
    		}
    	}
    }

![Center 1][]

三、Zookeeper实现分布式锁

分布式锁工具类

public class DistributedZookeepLock {
    
    	private static Logger logger = Logger.getLogger(DistributedZookeepLock.class.getName());
    	
    	public static final CuratorFramework client = CuratorFrameworkFactory.builder()
    			      .connectString("127.0.0.1:2181")
    									     .retryPolicy(new ExponentialBackoffRetry(1000,3)).build();
    	
    	
    	public static InterProcessMutex acquire(String lockPath) throws Exception{
    		if(client.getState() != CuratorFrameworkState.STARTED) {
    			client.start();
    		}
    		logger.log(Level.INFO, "加分布式锁:" + lockPath);
    		InterProcessMutex lock = new InterProcessMutex(client, lockPath);
    		lock.acquire();
    		logger.log(Level.INFO, "加分布式锁成功:" + lockPath);
    		return lock;
    	}
    	
    	public static void release(InterProcessMutex lock) throws Exception{
    		if(client.getState() != CuratorFrameworkState.STARTED){
    			System.out.println(Thread.currentThread().getName()+" release start");
    			client.start();
    		}
    		logger.log(Level.INFO, "释放分布式锁");
    		lock.release();
    		logger.log(Level.INFO, "释放分布式锁成功");
    	}
    
    }

public class LockTest {
    		private ReentrantLock lock = new ReentrantLock();
    		
    		public void inQueueWithScript(int value) {
    			System.out.println(Thread.currentThread().getName()+" inQueue");
    			long success = (long) JedisUtil.evalScript(LuaSysContants.IN_QUEUE.getName(), LuaSysContants.IN_QUEUE.getScript(), 1, "result_list", String.valueOf(value));
    			if(success == 0) {
    				System.err.println("入队出错！");
    			}
    		}
    		
    		public void inQueue(int value) {
    			// 分布式zookeeper锁
    			InterProcessMutex lock = null;
    			try {
    				lock = DistributedZookeepLock.acquire("/lock");
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    			try {
    				System.out.println(Thread.currentThread().getName()+" rpush");
    				long retValue = JedisUtil.rpush("result_list", String.valueOf(value));
    				System.out.println(Thread.currentThread().getName()+" llen");
    				long len = JedisUtil.llen("result_list");
    				if(retValue != len) {
    					System.err.println(retValue+":"+len);
    					System.err.println("入队出错！");
    				}
    			} finally {
    				try {
    					DistributedZookeepLock.release(lock);
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    		}
    		
    		public void outQueue() {
    			// 分布式zookeeper锁
    			InterProcessMutex lock = null;
    			try {
    				lock = DistributedZookeepLock.acquire("/lock");
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    			try {
    				System.out.println(Thread.currentThread().getName()+" lpop");
    				JedisUtil.lpop("result_list");
    			} finally {
    				try {
    					DistributedZookeepLock.release(lock);
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}

public static void main(String[] args) {
    		
    		CountDownLatch countDownLatch = new CountDownLatch(1);
    		
    		JedisUtil jedisUtil = new JedisUtil();
    		LockTest lockTest = new LockTest();
    		
    		Runnable inQueue = ()->{
    			for(int i=0; i<100; i++) {
    				try {
    					lockTest.inQueue(i);
    					countDownLatch.countDown();
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    		};
    		
    		Runnable outQueue = ()->{
    			try {
    				countDownLatch.await();
    				for(int i=0; i<30; i++) {
    					lockTest.outQueue();
    				}
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		};
    		
    		new Thread(inQueue).start();
    		new Thread(outQueue).start();
    		
    	}

![Center 2][]

[Center]: /images/20220606/1ac6cbc544684a1c899374f3673975b0.png
[Center 1]: /images/20220606/1169649243b84545b2795d9030283d1c.png
[Center 2]: /images/20220606/2d23aed1120a43e688c96f912f338032.png