基于redis分布式锁实现“秒杀”(代码片段)

CacheLockInterceptor实现InvocationHandler接口，在invoke方法中获取注解的方法和参数，在执行注解的方法前加锁，执行被注解的方法后释放锁：

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];

最关键的RedisLock类中的lock方法和unlock方法：

/**

• 加锁
• 使用方式为：
• 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 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);

模拟秒杀场景，1000个线程来争抢两个商品：

@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();
    

在正确的预想下，应该每个商品的库存都减少了500，在多次试验后，实际情况符合预想。如果不采用锁机制，会出现库存减少499，498的情况。
这里采用了动态代理的方法，利用注解和反射机制得到分布式锁ID，进行加锁和释放锁操作。当然也可以直接在方法进行这些操作，采用动态代理也是为了能够将锁操作代码集中在代理中，便于维护。
通常秒杀场景发生在web项目中，可以考虑利用spring的AOP特性将锁操作代码置于切面中，当然AOP本质上也是动态代理。

小结
这篇文章从业务场景出发，从抽象到实现阐述了如何利用redis实现分布式锁，完成简单的秒杀功能，也记录了笔者思考的过程，希望能给阅读到本篇文章的人一些启发。

基于golang实现redis分布式锁(代码片段)