基于Zookeeper的分布式共享锁

首先，说说我们的场景，订单服务是做成集群的，当两个以上结点同时收到一个相同订单的创建指令，这时并发就产生了，系统就会重复创建订单。等等......场景。这时，分布式共享锁就闪亮登场了。

共享锁在同一个进程中是很容易实现的，但在跨进程或者在不同Server之间就不好实现了。Zookeeper就很容易实现。具体的实现原理官网和其它网站也有翻译，这里就不在赘述了。

官网资料： http://zookeeper.apache.org/doc/r3.4.5/recipes.html

中文资料： https://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper

详见Locks章节。

原理都知道了，网上一搜索Apache上面已经有提供了，既然已经有轮子了，哪我们也没必要重复造轮子了吧！直接使用 Curator 。但是，我们在测试中发现，用于共享锁的结点无法自动回收，除了最末一级的临时结点会在锁释放和session超时的时候能自动回收外，其它结点均无法自动回收。我们的订单一天有好几万，遇到618和双十一的时候每天的订单量超50W，如果结点长期不回收的话，肯定会影响Zookeeper的性能。这时，我们就想到了一句话“自己动手，丰衣足食”。下面直接上代码：

首先，创建一个Maven工程，在pom文件里导入下面的包：

<dependencies>    <dependency>     <groupId>org.apache.zookeeper</groupId>     <artifactId>zookeeper</artifactId>     <version>3.4.6</version>    </dependency>    <dependency>     <groupId>org.apache.curator</groupId>     <artifactId>curator-client</artifactId>     <version>2.8.0</version>    </dependency>    <dependency>     <groupId>org.apache.curator</groupId>     <artifactId>curator-recipes</artifactId>     <version>2.8.0</version>    </dependency>    <dependency>     <groupId>org.apache.curator</groupId>     <artifactId>curator-framework</artifactId>     <version>2.8.0</version>    </dependency>    <dependency>     <groupId>commons-beanutils</groupId>     <artifactId>commons-beanutils</artifactId>     <version>1.9.2</version>    </dependency>    <dependency>     <groupId>commons-logging</groupId>     <artifactId>commons-logging</artifactId>     <version>1.2</version>    </dependency>    <dependency>     <groupId>commons-lang</groupId>     <artifactId>commons-lang</artifactId>     <version>2.6</version>    </dependency>   </dependencies>

LockZookeeperClient接口：

package com.XXX.framework.lock;    import org.apache.curator.framework.CuratorFramework;    /**   *    * description   *    * @author Roadrunners   * @version 1.0, 2015年7月9日   */  public interface LockZookeeperClient {   /**    *     * @return    */   CuratorFramework getCuratorFramework();     /**    *     * @return    */   String getBasePath();     /**    * garbage collector    *     * @param gcPath    */   void gc(String gcPath);  }

LockZookeeperClient接口的实现LockZookeeperClientFactory：

package com.XXX.framework.lock;    import java.util.Date;  import java.util.List;  import java.util.Timer;  import java.util.TimerTask;  import java.util.concurrent.ConcurrentSkipListSet;    import org.apache.commons.collections.CollectionUtils;  import org.apache.commons.lang.StringUtils;  import org.apache.commons.logging.Log;  import org.apache.commons.logging.LogFactory;  import org.apache.curator.framework.CuratorFramework;  import org.apache.curator.framework.CuratorFrameworkFactory;  import org.apache.curator.retry.ExponentialBackoffRetry;    /**   *    * description   *     * @author Roadrunners   * @version 1.0, 2015年7月9日   */  public class LockZookeeperClientFactory implements LockZookeeperClient {   private static final Log LOG = LogFactory.getLog(LockZookeeperClientFactory.class);     private boolean hasGc = true;   private Timer gcTimer;   private TimerTask gcTimerTask;   private ConcurrentSkipListSet<String> gcPaths = new ConcurrentSkipListSet<String>();   private int gcIntervalSecond = 60;     private CuratorFramework curatorFramework;   private String zookeeperIpPort = "localhost:2181";   private int sessionTimeoutMs = 10000;   private int connectionTimeoutMs = 10000;   private String basePath = "/locks";     public void setHasGc(boolean hasGc) {    this.hasGc = hasGc;   }     public void setGcIntervalSecond(int gcIntervalSecond) {    this.gcIntervalSecond = gcIntervalSecond;   }     public void setZookeeperIpPort(String zookeeperIpPort) {    this.zookeeperIpPort = zookeeperIpPort;   }     public void setSessionTimeoutMs(int sessionTimeoutMs) {    this.sessionTimeoutMs = sessionTimeoutMs;   }     public void setConnectionTimeoutMs(int connectionTimeoutMs) {    this.connectionTimeoutMs = connectionTimeoutMs;   }     public void setBasePath(String basePath) {    basePath = basePath.trim();    if (basePath.endsWith("/")) {     basePath = basePath.substring(0, basePath.length() - 1);    }      this.basePath = basePath;   }     public void init() {    if(StringUtils.isBlank(zookeeperIpPort)){     throw new NullPointerException("zookeeperIpPort");    }      ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);    curatorFramework = CuratorFrameworkFactory.newClient(zookeeperIpPort.trim(), sessionTimeoutMs, connectionTimeoutMs, retryPolicy);    curatorFramework.start();    LOG.info("CuratorFramework initialise succeed.");      if (hasGc) {     gc();    }   }     public void destroy() {    gcPaths.clear();    gcPaths = null;    gcStop();    curatorFramework.close();    curatorFramework = null;   }     @Override   public void gc(String gcPath) {    if (hasGc && StringUtils.isNotBlank(gcPath)) {     gcPaths.add(gcPath.trim());    }   }     @Override   public CuratorFramework getCuratorFramework() {    return this.curatorFramework;   }     @Override   public String getBasePath() {    return this.basePath;   }     private synchronized void gc() {    gcStop();      try {     scanningGCNodes();    } catch (Throwable e) {     LOG.warn(e);    }      gcTimerTask = new TimerTask() {     @Override     public void run() {      doingGc();     }    };      Date begin = new Date();    begin.setTime(begin.getTime() + (10 * 1000L));    gcTimer = new Timer("lock-gc", true);    gcTimer.schedule(gcTimerTask, begin, gcIntervalSecond * 1000L);   }     private synchronized void gcStop() {    if (null != gcTimer) {     gcTimer.cancel();     gcTimer = null;    }      if (null != gcTimerTask) {     gcTimerTask.cancel();     gcTimerTask = null;    }   }     private synchronized void scanningGCNodes() throws Exception {    if (null == curatorFramework.checkExists().forPath(basePath)) {     return;    }      List<String> paths = curatorFramework.getChildren().forPath(basePath);    if (CollectionUtils.isEmpty(paths)) {     gcPaths.add(basePath);     return;    }      for (String path : paths) {     try{      String tmpPath = basePath + "/" + path;      if (null == curatorFramework.checkExists().forPath(tmpPath)) {       continue;      }            gcPaths.add(tmpPath);     } catch(Throwable e){      LOG.warn("scanning gc nodes error.", e);     }    }   }      private synchronized void doingGc() {    LOG.debug("GC beginning.");      if (CollectionUtils.isNotEmpty(gcPaths)) {     for (String path : gcPaths) {      try {       if (null != curatorFramework.checkExists().forPath(path)) {        if (CollectionUtils.isEmpty(curatorFramework.getChildren().forPath(path))) {         curatorFramework.delete().forPath(path);         gcPaths.remove(path);         LOG.debug("GC " + path);        }       } else {        gcPaths.remove(path);       }      } catch (Throwable e) {       gcPaths.remove(path);       LOG.warn(e);      }     }    }      LOG.debug("GC ended.");   }    }

SharedLock共享锁：

package com.XXX.framework.lock.shared;    import java.util.concurrent.TimeUnit;    import org.apache.commons.lang.StringUtils;  import org.apache.curator.framework.recipes.locks.InterProcessLock;  import org.apache.curator.framework.recipes.locks.InterProcessMutex;    import com.XXX.framework.lock.LockZookeeperClient;    /**   *    * description   *    * @author Roadrunners   * @version 1.0, 2015年7月9日   */  public class SharedLock {   private InterProcessLock interProcessLock;     public SharedLock(LockZookeeperClient lockZookeeperClient, String resourceId) {    super();        if (StringUtils.isBlank(resourceId)) {     throw new NullPointerException("resourceId");    }    String path = lockZookeeperClient.getBasePath();    path += ("/" + resourceId.trim());      interProcessLock = new InterProcessMutex(lockZookeeperClient.getCuratorFramework(), path);    lockZookeeperClient.gc(path);   }         /**       * Acquire the mutex - blocking until it's available. Each call to acquire must be balanced by a call       * to {@link #release()}       *       * @throws Exception ZK errors, connection interruptions       */   public void acquire() throws Exception {    interProcessLock.acquire();   }        /**       * Acquire the mutex - blocks until it's available or the given time expires. Each call to acquire that returns true must be balanced by a call       * to {@link #release()}       *       * @param time time to wait       * @param unit time unit       * @return true if the mutex was acquired, false if not       * @throws Exception ZK errors, connection interruptions       */   public boolean acquire(long time, TimeUnit unit) throws Exception {    return interProcessLock.acquire(time, unit);   }        /**       * Perform one release of the mutex.       *       * @throws Exception ZK errors, interruptions, current thread does not own the lock       */   public void release() throws Exception {    interProcessLock.release();   }         /**       * Returns true if the mutex is acquired by a thread in this JVM       *       * @return true/false       */   public boolean isAcquiredInThisProcess() {    return interProcessLock.isAcquiredInThisProcess();   }  }

到此代码已经完成。下面写一个简单的Demo：

//LockZookeeperClientFactory通常是通过Spring配置注入的，此处是为了Demo的简单明了才这样写的，不建议这样写    LockZookeeperClientFactory lzc = new LockZookeeperClientFactory();    lzc.setZookeeperIpPort("10.100.15.1:8900");    lzc.setBasePath("/locks/sharedLock/");    lzc.init();      SharedLock sharedLock = new SharedLock(lzc, "sharedLock1");    try {     if (sharedLock.acquire(100, TimeUnit.MILLISECONDS)) {      System.out.println("sharedLock1 get");     }    } catch (Exception e) {     e.printStackTrace();    } finally {     try {      sharedLock.release();     } catch (Exception e) {      e.printStackTrace();     }    }      lzc.destroy();

就这样，系统就会每隔一分钟去回收一次没有使用的结点。