分布式锁框架

一 问题背景

1.1 问题背景

  看如下面这个例子中的请求情况。用户下单，检查库存，满足库存就下单成功，不满足库存就下单失败。

单节点部署在并发量很小的时候还是挺正常的,只要在库存检测的地方加上synchronized，整个流程的响应速度也算乐观，但是订单系统或库存系统其中任意一台服务down掉，都会中断整个业务流程。（耦合度过高，存在单点故障）。因此才决定要改用分布式集群部署方案解决单点故障，提高系统可用性。

 

分布式环境下

改用分布式集群部署方案后的流程图如上所示，看起来只是将订单系统与库存系统多部署了几个节点而已，但是这样在整个业务流程上却有所不同，首先我们先来分析一下此方案为什么会出现线程安全问题。场景如下

 

场景一: 小A、小B同时向订单系统进行下单，由于订单系统的负载均衡策略，将小何的请求交给了订单系统1去处理，小黄的请求交给了订单系统2去处理

订单系统1查询了库存系统1，发现库存充足

订单系统2查询了库存系统2，发现库存充足

订单系统1、订单系统2都发现库存充足，于是分别通知库存系统1、库存系统2进行扣减库存，这就导致了库存系统中商品数量被扣减为负数的情况，也叫做库存超卖现象。

  

场景二: 还有一种情况是小A同学，由于在点击下单按钮的时候，由于网络稍有延迟，导致小A同学在点击下单的时候，多点了几次，刚好这两个点击请求分发到了不同的服务器实例上。导致，产生了同一个用户的并发问题。

出现这个问题的根本原因是分布式集群节点之间无法共享synchronized锁。既然问题已经分析出来了，那么解决问题方案自然就呼之欲出了。我们可以使用一把能够跨应用共享的分布式锁，锁住扣减库存的过程，这样一来在订单系统扣减库存的过程中，就不允许有其他的订单系统执行相同的操作。这样就可以保证了分布式集群环境下的线程安全性问题。

以上描述很容易看出来，就是分布式多实例部署之下引发的并发问题。那么就要求我们在空间的基础上能够把这个并发的问题解决了

1.2 为什么要分布式锁

1 怎么保证分布式系统中唯一键的生成

2 分布式环境下怎么保证，在高并发的情况下数据的唯一性问题

3 分布式环境下怎么保证数据并发写入的时候 不重复的问题

二 分布式锁应用

场景描述:

1 只要是首次玩家，进入答题详情页就给用户发一张复活卡，一张延迟卡，然而这里有一个问题就是先查出有没有发过count然后才发。这个过程如果多实例并发就有肯能导致，新玩家卡重复发送。所以这里加锁，键值为可以控制卡片唯一性的主键，也就是uid+hdid+卡片来源，保证了唯一性.

 

 

2成语答题场景

领取额外奖励的场景,再答n题领取奖励，领取额外奖励每个用户只能领取到一次，这个时候如果并发量比较大，重复点击领取同一个任务奖励。就会出现不同一个用户领取任务的奖励分发到不同的实例上。由此可能产生并发问题。

 

如上代码所示，就是领取任务的代码，然而这个过程就有可能出现同一个用户由于并发领取任务导致重复加金币的问题。这里比如上面红框部分我查询出配置，两个请求并发的来，我不加锁，并发查询的话，可能都没有查到就有可能同时写入数据库并且同时加金币。这个过程中就有必要加一把分布式的锁。保证这个业务的唯一性。但是这个过程中就有一个问题了如果这两个并发的请求都是有效的呢，如果一个有效另一个无效呢。这个过程中我并不知道两个请求是不是都是有效的。那么并发控制的键就成为了关键。这里问题就在与并发要控制的公共资源是什么了。所以我在这里主要是要控制同一个用户不能并发领取同一个奖励。好了那这个键就变成了锁住用户id和任务id了。

综上所请求的，并发问题总结

1. 凡是要保证数据唯一性的地方检查并发问题。比如新玩家的卡一个用户只能有一张新玩家复活卡，或者延迟卡，那么就要考虑是不是数据库层面要加唯一索引解决并发问题。同时考虑代码层面是不是需要加分布式锁。比如一个用户只能领取一个任务uid,task_id其实也是可以做唯一索引的，这个时候就可以考虑是不是有并发问题是不是要加分布式锁。
2. 凡是涉及到数据库层查询与修改的原子性问题的，就是有查询之后根据查询结果修改数据库更新操作的，需要原子性的代码。比如查询数据库中有没有，没有就插入数据这样的代码，首先是保证代码快的原子性，可能在一个实例查的时候另一个实例也再查询，如果没有实现不保证代码快的空间原子性问题，就可能导致并发问题。

 

3 如上所列举的这些问题，在分布式环境下都是可以分布式锁来解决的，关键在于分布式锁要怎么实现呢

这里就有好几种实现方案了

    1.使用数据库实现分布式锁

     缺点:性能差、线程出现异常时，容易出现死锁

    2.使用redis实现分布式锁

      缺点:锁的失效时间难控制、容易产生死锁、非阻塞式、不可重入

       1 当执行的时候抛异常了 (异常以后释放锁)
       2 获得锁以后执行方法的时候 当前机器宕机了导致无法释放(设置redis中key过期时间让其自动释放锁)
       3 设置了持有锁的时间以后 释放了别的机器的锁（可以在本地生成随机码，作为redis的值 每一次加锁的时候获取锁）
       4 设置了值以后 get(key)==value doSomeThing 如果不是原子性的会出现并发问题，解决把办法lua脚本

   使用redis实现分布式锁，要处理的异常问题太多，不易于处理相关问题。

   3.使用zookeeper实现分布式锁

    实现相对简单、可靠性强、使用临时节点，失效时间容易控制

 

   如上图所示我们在zookeeper上挂载一个节点，通过不同的注解在调用同步代码的时候创建临时节点来实现分布式缩。如上图所示 当我们的一台机进入同步代码的时候就在zookeeper上创建一个临时节点，同步代码块走完以后就释放临时节点。这样就可以实现分布式锁的效果了。

1 抽象锁的实现

public interface Lock {
   public void lock();
   public void unlock();
}

2 子类

/**
 * @author Administrator
 * 抽象的zookeeper锁
 */
public abstract class AbstractZookeeperLock implements Lock{
 
	private static final String address="127.0.0.1:2181";
	
	protected  ZkClient zkClient=new ZkClient(address);
	
	protected  String PATH="/lock";
	
	protected CountDownLatch countDownLatch=null;
 
	
	public void lock() {
	       if(tryLock()){
	    	   System.out.println("获取到锁");
	       }else{
	    	   //如果没有获取到锁的时候等待，被唤醒以后再次尝试获取锁
	    	   await();
	    	   //再次尝试获取锁
	    	   lock();
	       }
		
	}
 
	protected abstract void await();
 
	protected abstract boolean tryLock();
 
	public void unlock() {
		if(zkClient!=null){
			
		}
		zkClient.close();
	}
 
}

 

3 子类修饰

public class ZookeeperDistrabuteLock  extends AbstractZookeeperLock {
 
	@Override
	protected void await() {
		IZkDataListener zkDataListener= new IZkDataListener() {
			public void handleDataDeleted(String path) throws Exception {
				countDownLatch.countDown();
			}
			public void handleDataChange(String arg0, Object arg1) throws Exception {
				
			}
		};
	   //注册节点的舰艇时间 
       zkClient.subscribeDataChanges(PATH, zkDataListener);		
       //如果节点存在就等待
	   if(zkClient.exists(PATH)){
			countDownLatch=new CountDownLatch(1);
			try {
				countDownLatch.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
	    }
		//如果节点删除以后，就可以取消监听 后面尝试去获取锁了
		zkClient.unsubscribeDataChanges(PATH, zkDataListener);
		
	}
 
	@Override
	protected boolean tryLock() {
		try {
			zkClient.createEphemeral(PATH);
			return true;
		} catch (Exception e) {
			return false;
		}
	}
 
}

4 通步代码块

public class IdGenerateTools {
    public static long generateId(){
    	long currentTimeMillis = System.currentTimeMillis();
    	return currentTimeMillis;
    }
}

5 测试类(模拟多台计算机线程)

public class TestUserServiceIm {
	private static ExecutorService executorService=Executors.newCachedThreadPool();
	public static void main(String[] args) {
		for(int i=0;i<100;i++){
		   executorService.execute(new UserServiceIm());
		}
	}
}

6 使用zookeeper分布式锁

/**
 * @author Administrator
 * 用户信息服务实现类
 */
public class UserServiceIm implements Runnable {
 
	ZookeeperDistrabuteLock distrabuteLock=new ZookeeperDistrabuteLock();
	
	public void run() {
		try {
			distrabuteLock.lock();
			long generateId = IdGenerateTools.generateId();
			Thread.sleep(1000);
			System.out.println(Thread.currentThread().getName()+" 订单唯一Id "+generateId);
		} catch (Exception e) {
			// TODO: handle exception
		}finally {
			distrabuteLock.unlock();
		}
		
	}
	
}

如上就是分布式锁的实现方式。

         大家可以看到这个是是基于zookeeper实现的分布式锁。然而让你觉得有点不方便的事情又来了。这个锁都是在方法执行前获取到锁。方法执行以后释放掉锁资源的。那么这个跟什么有点像呢。正如你所看到的这跟我们的代理有点像。

那么基于这个思想我们就可以使用AOP+注解的方式。来实现在方法上加注解。通过AOP编程来实现分布式锁。这样以后用的时候就可以直接打个注解。就可以使用分布式锁了。

 

  4.使用redis实现分布式锁

1 用redis的setnx来设置值，设置成功以后执行代码，执行完以后finally中删除key

2 由于有程序中断的可能，所以setnx的时候要加 超时时间

3 为了保证，超时时间不在代码没有执行完之前过期，所以每一次有线程请求没有成功在时候，都要把过期时间1/3，这样才能保证执行完

4 为了保证，一个线程在删除的时候，确确实实删除的是自己的锁，那么在设置分布式锁的时候，最好给线程设置一个UUID的值，释放时候

先取值判断一下是不是当前线程。

5为了保证同一个线程可以重入，最好引入threadlocal把，当前线程的uuid管理起来，如果能get出来就直接获得锁。

6 为了保证一个线程在执行的时候是锁等待，而不是丢失请求，这里我们可以引入Redission。