基于redis分布式鎖實(shí)現(xiàn)秒殺功能

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface CacheLock {
 String lockedPrefix() default "";//redis 鎖key的前綴
 long timeOut() default 2000;//輪詢鎖的時(shí)間
 int expireTime() default 1000;//key在redis里存在的時(shí)間，1000S
}

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

@Target(ElementType.PARAMETER)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface LockedComplexObject {
 String field() default "";//含有成員變量的復(fù)雜對(duì)象中需要加鎖的成員變量，如一個(gè)商品對(duì)象的商品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);
 }
 //獲得方法中參數(shù)的注解
 Annotation[][] annotations = method.getParameterAnnotations();
 //根據(jù)獲取到的參數(shù)注解和參數(shù)列表獲得加鎖的參數(shù)
 Object lockedObject = getLockedObject(annotations,args);
 String objectValue = lockedObject.toString();
 //新建一個(gè)鎖
 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{
 //加鎖成功，執(zhí)行方法
 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("方法參數(shù)為空，沒有被鎖定的對(duì)象");
 }

 if(null == annotations || annotations.length == 0){
 throw new CacheLockException("沒有被注解的參數(shù)");
 }
 //不支持多個(gè)參數(shù)加鎖，只支持第一個(gè)注解為lockedObject或者lockedComplexObject的參數(shù)
 int index = -1;//標(biāo)記參數(shù)的位置指針
 for(int i = 0;i  annotations.length;i++){
 for(int j = 0;j  annotations[i].length;j++){
 if(annotations[i][j] instanceof LockedComplexObject){//注解為L(zhǎng)ockedComplexObject
  index = i;
  try {
  return args[i].getClass().getField(((LockedComplexObject)annotations[i][j]).field());
  } catch (NoSuchFieldException | SecurityException e) {
  throw new CacheLockException("注解對(duì)象中沒有該屬性" + ((LockedComplexObject)annotations[i][j]).field());
  }
 }

 if(annotations[i][j] instanceof LockedObject){
  index = i;
  break;
 }
 }
 //找到第一個(gè)后直接break，不支持多參數(shù)加鎖
 if(index != -1){
 break;
 }
 }

 if(index == -1){
 throw new CacheLockException("請(qǐng)指定被鎖定參數(shù)");
 }

 return args[index];
 }
}

/**
 * 加鎖
 * 使用方式為：
 * lock();
 * try{
 * executeMethod();
 * }finally{
 * unlock();
 * }
 * @param timeout timeout的時(shí)間范圍內(nèi)輪詢鎖
 * @param expire 設(shè)置鎖超時(shí)時(shí)間
 * @return 成功 or 失敗
 */
 public boolean lock(long timeout,int expire){
 long nanoTime = System.nanoTime();
 timeout *= MILLI_NANO_TIME;
 try {
 //在timeout的時(shí)間范圍內(nèi)不斷輪詢鎖
 while (System.nanoTime() - nanoTime  timeout) {
 //鎖不存在的話，設(shè)置鎖并設(shè)置鎖過(guò)期時(shí)間，即加鎖
 if (this.redisClient.setnx(this.key, LOCKED) == 1) {
  this.redisClient.expire(key, expire);//設(shè)置鎖過(guò)期時(shí)間是為了在沒有釋放
  //鎖的情況下鎖過(guò)期后消失，不會(huì)造成永久阻塞
  this.lock = true;
  return this.lock;
 }
 System.out.println("出現(xiàn)鎖等待");
 //短暫休眠，避免可能的活鎖
 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 {
/**
*現(xiàn)在暫時(shí)只支持在接口方法上注解
*/
 //cacheLock注解可能產(chǎn)生并發(fā)的方法
 @CacheLock(lockedPrefix="TEST_PREFIX")
 public void secKill(String userID,@LockedObject Long commidityID);//最簡(jiǎn)單的秒殺方法，參數(shù)是用戶ID和商品ID。可能有多個(gè)線程爭(zhēng)搶一個(gè)商品，所以商品ID加上LockedObject注解
}

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

 @Override
 public void secKill(String arg1, Long arg2) {
 //最簡(jiǎn)單的秒殺，這里僅作為demo示例
 reduceInventory(arg2);
 }
 //模擬秒殺操作，姑且認(rèn)為一個(gè)秒殺就是將庫(kù)存減一，實(shí)際情景要復(fù)雜的多
 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個(gè)線程，秒殺第一個(gè)商品
 for(int i= 0;i  splitPoint;i++){
 threads[i] = new Thread(new Runnable() {
 public void run() {
  try {
  //等待在一個(gè)信號(hào)量上，掛起
  beginCount.await();
  //用動(dòng)態(tài)代理的方式調(diào)用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個(gè)線程，秒殺第二件商品
 for(int i= splitPoint;i  threadCount;i++){
 threads[i] = new Thread(new Runnable() {
 public void run() {
  try {
  //等待在一個(gè)信號(hào)量上，掛起
  beginCount.await();
  //用動(dòng)態(tài)代理的方式調(diào)用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();
 //主線程釋放開始信號(hào)量，并等待結(jié)束信號(hào)量，這樣做保證1000個(gè)線程做到完全同時(shí)執(zhí)行，保證測(cè)試的正確性
 beginCount.countDown();

 try {
 //主線程等待結(jié)束信號(hào)量
 endCount.await();
 //觀察秒殺結(jié)果是否正確
 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();
 }
 }