基于redis分布式锁实现秒杀功能

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface CacheLock {
 String lockedPrefix() default "";//redis 锁key的前缀
 long timeOut() default 2000;//轮询锁的时间
 int expireTime() default 1000;//key在redis里存在的时间，1000S
}

@Target(ElementType.PARAMETER)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface LockedObject {
 //不需要值
}

@Target(ElementType.PARAMETER)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface LockedComplexObject {
 String field() default "";//含有成员变量的复杂对象中需要加锁的成员变量，如一个商品对象的商品ID

}

public class CacheLockInterceptor implements InvocationHandler{
 public static int ERROR_COUNT = 0;
 private Object proxied;

 public CacheLockInterceptor(Object proxied) {
 this.proxied = proxied;
 }

 @Override
 public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

 CacheLock cacheLock = method.getAnnotation(CacheLock.class);
 //没有cacheLock注解，pass
 if(null == cacheLock){
  System.out.println("no cacheLock annotation");  
  return method.invoke(proxied, args);
 }
 //获得方法中参数的注解
 Annotation[][] annotations = method.getParameterAnnotations();
 //根据获取到的参数注解和参数列表获得加锁的参数
 Object lockedObject = getLockedObject(annotations,args);
 String objectValue = lockedObject.toString();
 //新建一个锁
 RedisLock lock = new RedisLock(cacheLock.lockedPrefix(), objectValue);
 //加锁
 boolean result = lock.lock(cacheLock.timeOut(), cacheLock.expireTime());
 if(!result){//取锁失败
  ERROR_COUNT += 1;
  throw new CacheLockException("get lock fail");

 }
 try{
  //加锁成功，执行方法
  return method.invoke(proxied, args);
 }finally{
  lock.unlock();//释放锁
 }

 }
 /**
 * 
 * @param annotations
 * @param args
 * @return
 * @throws CacheLockException
 */
 private Object getLockedObject(Annotation[][] annotations,Object[] args) throws CacheLockException{
 if(null == args || args.length == 0){
  throw new CacheLockException("方法参数为空，没有被锁定的对象");
 }

 if(null == annotations || annotations.length == 0){
  throw new CacheLockException("没有被注解的参数");
 }
 //不支持多个参数加锁，只支持第一个注解为lockedObject或者lockedComplexObject的参数
 int index = -1;//标记参数的位置指针
 for(int i = 0;i < annotations.length;i++){
  for(int j = 0;j < annotations[i].length;j++){
  if(annotations[i][j] instanceof LockedComplexObject){//注解为LockedComplexObject
   index = i;
   try {
   return args[i].getClass().getField(((LockedComplexObject)annotations[i][j]).field());
   } catch (NoSuchFieldException | SecurityException e) {
   throw new CacheLockException("注解对象中没有该属性" + ((LockedComplexObject)annotations[i][j]).field());
   }
  }

  if(annotations[i][j] instanceof LockedObject){
   index = i;
   break;
  }
  }
  //找到第一个后直接break，不支持多参数加锁
  if(index != -1){
  break;
  }
 }

 if(index == -1){
  throw new CacheLockException("请指定被锁定参数");
 }

 return args[index];
 }
}

/**
 * 加锁
 * 使用方式为：
 * lock();
 * try{
 * executeMethod();
 * }finally{
 * unlock();
 * }
 * @param timeout timeout的时间范围内轮询锁
 * @param expire 设置锁超时时间
 * @return 成功 or 失败
 */
 public boolean lock(long timeout,int expire){
 long nanoTime = System.nanoTime();
 timeout *= MILLI_NANO_TIME;
 try {
  //在timeout的时间范围内不断轮询锁
  while (System.nanoTime() - nanoTime < timeout) {
  //锁不存在的话，设置锁并设置锁过期时间，即加锁
  if (this.redisClient.setnx(this.key, LOCKED) == 1) {
   this.redisClient.expire(key, expire);//设置锁过期时间是为了在没有释放
   //锁的情况下锁过期后消失，不会造成永久阻塞
   this.lock = true;
   return this.lock;
  }
  System.out.println("出现锁等待");
  //短暂休眠，避免可能的活锁
  Thread.sleep(3, RANDOM.nextInt(30));
  } 
 } catch (Exception e) {
  throw new RuntimeException("locking error",e);
 }
 return false;
 }

 public void unlock() {
 try {
  if(this.lock){
  redisClient.delKey(key);//直接删除
  }
 } catch (Throwable e) {

 }
 }

public interface SeckillInterface {
/**
*现在暂时只支持在接口方法上注解
*/
 //cacheLock注解可能产生并发的方法
 @CacheLock(lockedPrefix="TEST_PREFIX")
 public void secKill(String userID,@LockedObject Long commidityID);//最简单的秒杀方法，参数是用户ID和商品ID。可能有多个线程争抢一个商品，所以商品ID加上LockedObject注解
}

public class SecKillImpl implements SeckillInterface{
 static Map<Long, Long> inventory ;
 static{
 inventory = new HashMap<>();
 inventory.put(10000001L, 10000l);
 inventory.put(10000002L, 10000l);
 }

 @Override
 public void secKill(String arg1, Long arg2) {
 //最简单的秒杀，这里仅作为demo示例
 reduceInventory(arg2);
 }
 //模拟秒杀操作，姑且认为一个秒杀就是将库存减一，实际情景要复杂的多
 public Long reduceInventory(Long commodityId){
 inventory.put(commodityId,inventory.get(commodityId) - 1);
 return inventory.get(commodityId);
 }

}

@Test
 public void testSecKill(){
 int threadCount = 1000;
 int splitPoint = 500;
 CountDownLatch endCount = new CountDownLatch(threadCount);
 CountDownLatch beginCount = new CountDownLatch(1);
 SecKillImpl testClass = new SecKillImpl();

 Thread[] threads = new Thread[threadCount];
 //起500个线程，秒杀第一个商品
 for(int i= 0;i < splitPoint;i++){
  threads[i] = new Thread(new Runnable() {
  public void run() {
   try {
   //等待在一个信号量上，挂起
   beginCount.await();
   //用动态代理的方式调用secKill方法
   SeckillInterface proxy = (SeckillInterface) Proxy.newProxyInstance(SeckillInterface.class.getClassLoader(), 
    new Class[]{SeckillInterface.class}, new CacheLockInterceptor(testClass));
   proxy.secKill("test", commidityId1);
   endCount.countDown();
   } catch (InterruptedException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
   }
  }
  });
  threads[i].start();

 }
 //再起500个线程，秒杀第二件商品
 for(int i= splitPoint;i < threadCount;i++){
  threads[i] = new Thread(new Runnable() {
  public void run() {
   try {
   //等待在一个信号量上，挂起
   beginCount.await();
   //用动态代理的方式调用secKill方法
   SeckillInterface proxy = (SeckillInterface) Proxy.newProxyInstance(SeckillInterface.class.getClassLoader(), 
    new Class[]{SeckillInterface.class}, new CacheLockInterceptor(testClass));
   proxy.secKill("test", commidityId2);
   //testClass.testFunc("test", 10000001L);
   endCount.countDown();
   } catch (InterruptedException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
   }
  }
  });
  threads[i].start();

 }


 long startTime = System.currentTimeMillis();
 //主线程释放开始信号量，并等待结束信号量，这样做保证1000个线程做到完全同时执行，保证测试的正确性
 beginCount.countDown();

 try {
  //主线程等待结束信号量
  endCount.await();
  //观察秒杀结果是否正确
  System.out.println(SecKillImpl.inventory.get(commidityId1));
  System.out.println(SecKillImpl.inventory.get(commidityId2));
  System.out.println("error count" + CacheLockInterceptor.ERROR_COUNT);
  System.out.println("total cost " + (System.currentTimeMillis() - startTime));
 } catch (InterruptedException e) {
  // TODO Auto-generated catch block
  e.printStackTrace();
 }
 }