基于Zookeeper的锁开发手册.docx

基于Zookeeper的锁应用开发手册基于Zookeeper的锁应用开发手册版本 修订历史日期版本描述修改人2013-10-251.0基于Zookeeper的锁应用开发手册苏晓辉基于Zookeeper的锁应用开发手册41 概述41.1前言41.2目的41.3范围41.4术语和缩写语42 排他锁53 共享锁64 数据库锁75 总结和展望86 附件代码97 参考资料36基于Zookeeper的锁应用开发手册1 概述1.1 前言Zookeeper是Hadoop的正式子项目，它是一个针对大型分布式系统的可靠协调系统，提供的功能包括：配置维护、名字服务、分布式同步、组服务等。Zookeeper的目标就是封装好复杂易出错的关键服务，将简单易用的接口和性能高效、功能稳定的系统提供给用户。1.2 目的本文档的目的是帮助PAAS平台开发人员快速使用锁API，开发分布式系统中模块之间需要协同执行的相关的应用程序。本文第二、三、四章分别介绍排他锁、共享锁和数据库锁的功能和使用。最后，第五、六、七章给出附件代码、总结及展望和参考资料。1.3 范围本文档适用于PAAS系统整个项目周期中对模块之间需要协同执行的应用程序开发的指导和约束。要了解Zookeeper的使用详情，参考Zookeeper3.4.5官方文档。1.4 术语和缩写语术语说明ZookeeperZookeeper是Hadoop的正式子项目，它是一个针对大型分布式系统的可靠协调系统。PAASPlatform-as-a-Service的缩写，意思是平台即服务。 把服务器平台作为一种服务提供的商业模式。SAASsoftware-as-a-Service的缩写。国外称为SAAS，国内通常叫做软件运营服务模式，简称为软营模式。2 排他锁排他锁的实现机制是当一个线程获取锁后，另一个线程无法获取该锁，并报出异常，这里的线程可以在不同的JVM中；目前支持重入特性，在同一线程中的不同地方（不同的方法中）可以重复获取同一把锁。排他锁的使用场景是当一个线程需要对分布式系统中的某些资源独占，不希望其它线程对该资源进行操作。排他锁使用范例如下代码所示，如果代码lock.lock()获取锁key1锁，则进入独占资源操作代码，否则进入异常处理代码，最后释放锁lock.close()。MutexLock lock = null;try lock = new MutexLockImplZooKeeper(:2181, key1); lock.lock();/独占资源操作代码 catch (LockException e) /异常处理代码 finally try if(lock!=null) lock.close(); catch (LockException e) 3 共享锁共享锁有读锁和写锁两种状态，共享锁的实现机制是当一个线程获取锁的读状态后，另一个线程也可以获取该锁的读状态，但是另一个线程无法获取该锁的写状态，并抛出异常；当一个线程获取锁的写状态后，另一个线程无法获取该锁，并抛出异常。这里的线程可以在不同的JVM中；目前支持重入特性，在同一线程中的不同地方（不同的方法中）可以重复获取同一把锁。共享锁的使用场景是当一个线程需要对分布式系统中的某些资源进行读取操作时，其它线程也可以读取该资源，但是不能进行写操作；当一个线程需要对分布式系统中的某些资源进行写操作时，不希望其它线程对该资源进行操作。共享锁使用范例如下代码所示，如果代码lock.read()/lock.write()获取锁key1锁，则进入独占资源操作代码，否则进入异常处理代码，最后释放锁lock.close()。SharedLocklock = null;try lock = new SharedLockImplZooKeeper (:2181, key1); lock.read();/lock.write();/独占资源操作代码 catch (LockException e) /异常处理代码 finally try if(lock!=null) lock.close(); catch (LockException e) 4 数据库锁目前数据库锁支持4种标准的数据库隔离等级，包括Read Uncommit、Read Commit、No Repeating和Serialize，目前不支持锁的重入功能。数据库锁的使用场景是需要提供数据库事务ACID特性的地方。数据库锁使用范例如下代码所示，如果代码readLock(group1, table1)/ writeLock(group1, table1, ID)获取数据库表锁或行锁，否则进入异常处理代码，最后释放锁release()。DBLockImplZooKeeper lock = null;try lock = new DBLockImplZooKeeper(:2181); /or new DBLockImplZooKeeper(:2181, IsolateLevel.READCOMMIT);/or new DBLockImplZooKeeper(:2181, 3, 100, 10000, IsolateLevel.READCOMMIT);l.readLock(group1, table1);/操作代码l.writeLock(group1, table1, ID);/操作代码 catch (LockException e) /异常处理代码 finally try if(lock!=null) l.release(); catch (LockException e) 5 总结和展望将来可能还需要实现两阶段关闭（two phase temiate）锁来协同多个模块完成同一个工作，最后各模块要么一起成功，要么一起失败。6 附件代码SharedLock.java /(c) Copyright Beyond PAAS contributors and others 2013package com.beyond.paas.tool.lock;/* * SharedLock is interface. * author su xiaohui * CreateDate 2013-10-18 */public interface SharedLock /* * To get write lock * throws LockException */void write()throws LockException;/* * To get read lock * throws LockException */void read()throws LockException;/* * To release lock * throws LockException */void close() throws LockException;SharedLockImplZooKeeper.java/(c) Copyright Beyond PAAS contributors and others 2013package com.beyond.paas.tool.lock;import java.io.IOException;import java.util.ArrayList;import java.util.Collections;import java.util.Comparator;import java.util.HashSet;import java.util.List;import java.util.Set;import java.util.UUID;import org.apache.zookeeper.CreateMode;import org.apache.zookeeper.KeeperException;import org.apache.zookeeper.ZooKeeper;import org.apache.zookeeper.ZooDefs.Ids;/* * SharedLockImplZooKeeper is threadSafed, so the class cant be extended and modified. On the other hand, * private and final and synchronized keywords cant be remove and modified by other keywords except to * you are sure that the class is threadSafed. * SharedLockImplZooKeeper provide read-write lock function and support locks reenter function. * SharedLockImplZooKeeper depends on the zookeeper severs, so you must be sure that zookeeper servers * are work. * usage: SharedLock l = null; try l = new SharedLock(:2181, key1); /or new SharedLock(:2181, key1, 3, 100, 10000);.l.write();/or l.read();. catch (LockException e) . finally try if(l!=null) l.close(); catch (LockException e) * author su xiaohui * CreateDate 2013-10-18 */final public class SharedLockImplZooKeeper implements SharedLockfinal private String root = /shared;final private String serverAddress;final private String ownID;final private int retry;final private int waitTime;final private int timeout;private boolean isReentry;private int count;private int step;private ZooKeeper zk;final private String zookeeperID;final static private ThreadLocalSet session = new ThreadLocalSet(); /* * Construction * param serverAddress. Zookeeper servers address-port like :2181,:2181,. * param ownID. Source Key or lock name. * throws LockException */public SharedLockImplZooKeeper(String serverAddress, String ownID) throws LockException this(serverAddress, ownID, 0, 100, 10000);/* * Construction * param serverAddress. Zookeeper servers address-port like :2181,:2181,. * param ownID. Source Key or lock name. * param retry. Retry times. * param waitTime. Wait time. * param timeout. Time out. * throws LockException */public SharedLockImplZooKeeper(String serverAddress, String ownID, int retry, int waitTime, int timeout) throws LockExceptionthis.serverAddress = serverAddress;this.ownID = ownID;this.retry = retry;this.waitTime = waitTime;this.timeout = timeout;this.step = 1;this.count = 0;this.zookeeperID = UUID.randomUUID().toString();private void init(String type) throws LockExceptionif(type.equals(read)if(session.get()=null | (!session.get().contains(ownID+_write) & !session.get().contains(ownID+_read) )connectZooKeeper();Set set = session.get()=null ? new HashSet() : session.get();set.add(ownID+_read);session.set(set);isReentry = false;elseisReentry = true;else if(type.equals(write)if(session.get()=null | !session.get().contains(ownID+_write) )if(session.get()!=null & session.get().contains(ownID+_read)throw new LockException(new Exception(The lock has been locked.);connectZooKeeper();Set set = session.get()=null ? new HashSet() : session.get();set.add(ownID+_write);session.set(set);isReentry = false;elseisReentry = true;Overridesynchronized public void write() throws LockException if(step!=1)throw new LockException(new Exception(This lock is invalid.);elsestep = 2;init(write);while(true & !isReentry)count+;tryexecuteWrite();break;catch(LockException ex)if(count=retry)closeLocalThread();closeZookeeper();throw new LockException(ex);elsetryThread.sleep(waitTime);catch(Exception e)private void executeWrite() throws LockExceptionString path = createPath();String actualPath = ;try actualPath = zk.create(path+/+ownID+#+zookeeperID+#write#, new byte0, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);List list1 = zk.getChildren(path, null);List list2 = new ArrayList();for(String node : list1)if(node.split(#).length=4) list2.add(node);Collections.sort(list2,new Comparator()Overridepublic int compare(String node1, String node2) String node1s = node1.split(#);String node2s = node2.split(#);if(Long.parseLong(node1s3)Long.parseLong(node2s3)return 1;if(Long.parseLong(node1s3)-1)if(node.indexOf(zookeeperID)=-1)throw new LockException(new Exception(The lock has been locked.);elsebreak; catch (KeeperException e) throw new LockException(e); catch (InterruptedException e) throw new LockException(e); catch (LockException e) if(actualPath!=null)try zk.delete(actualPath, 0); catch (InterruptedException e1) catch (KeeperException e1) throw e;Overridepublic void read() throws LockException if(step!=1)throw new LockException(new Exception(This lock is invalid.);elsestep = 2;init(read);while(true & !isReentry)count+;tryexecuteRead();break;catch(LockException ex)if(count=retry)closeLocalThread();closeZookeeper();throw new LockException(ex);elsetryThread.sleep(waitTime);catch(Exception e)private void executeRead() throws LockException String path = createPath();String actualPath = ;try actualPath = zk.create(path+/+ownID+#+zookeeperID+#read#, new byte0, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);List list1 = zk.getChildren(path, null);List list2 = new ArrayList();for(String node : list1)if(node.split(#).length=4) list2.add(node);Collections.sort(list2,new Comparator()Overridepublic int compare(String node1, String node2) String node1s = node1.split(#);String node2s = node2.split(#);if(Long.parseLong(node1s3)Long.parseLong(node2s3)return 1;if(Long.parseLong(node1s3)-1)if(node.indexOf(zookeeperID)=-1)if(node.indexOf(#write#)-1) throw new LockException(new Exception(The lock has been write locked.);elsebreak; catch (KeeperException e) throw new LockException(e); catch (InterruptedException e) throw new LockException(e); catch (LockException e) if(actualPath!=null)try zk.delete(actualPath, -1); catch (InterruptedException e1) catch (KeeperException e1) throw e;private String createPath()tryif(zk.exists(root, false)=null)zk.create(root, new byte0, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); catch (KeeperException e) catch (InterruptedException e) return root;private void connectZooKeeper()throws LockException try if(zk=null) zk = new ZooKeeper(serverAddress, timeout, null); catch (IOException e) throw new LockException(e);private void closeLocalThread()if(!isReentry & session.get()!=null)session.get().remove(ownID+_write);session.get().remove(ownID+_read);private void closeZookeeper()if(zk!=null)try zk.close(); catch (InterruptedException e) finallyzk = null;Overridesynchronized public void close() throws LockException if(step!=2)throw new LockException(new Exception(The lock is invalid);elsestep = 3;closeLocalThread();closeZookeeper();MutexLock.java/(c) Copyright Beyond PAAS contributors and others 2013package com.beyond.paas.tool.lock;/* * MutexLock is interface. * author su xiaohui * CreateDate 2013-10-18 */public interface MutexLock /* * To get lock * throws LockException */void lock() throws LockException;/* * To release lock * throws LockException */void close() throws LockException; MutexLockImplZooKeeper.java/(c) Copyright Beyond PAAS contributors and others 2013package com.beyond.paas.tool.lock;import java.io.IOException;import java.util.HashSet;import java.util.Set;import org.apache.zookeeper.CreateMode;import org.apache.zookeeper.KeeperException;import org.apache.zookeeper.ZooKeeper;import org.apache.zookeeper.ZooDefs.Ids;/* * MutexLockImplZooKeeper is threadSafed, so the class cant be extended and modified. On the other hand, * private and final and synchronized keywords cant be remove and modified by other keywords except to * you are sure that the class is threadSafed. * MutexLockImplZooKeeper provide mutual lock function and support locks reenter function. * MutexLockImplZooKeeper depends on the zookeeper severs, so you must be sure that zookeeper servers * are work. * usage: MutexLock l = null; try l = new MutexLockImplZooKeeper(:2181, key1); /or l = new MutexLockImplZooKeeper(:2181, key1, 3, 100, 10000);.l.lock();. catch (LockException e) . finally try if(l!=null) l.close(); catch (LockException e) * author su xiaohui * CreateDate 2013-10-18 */final public class MutexLockImplZooKeeper implements MutexLockfinal private String root = /mutex;final private String serverAddress;final private String ownID;final private int retry;final private int waitTime;final private int timeout;private boolean isReentry;private int count;private int step;private ZooKeeper zk;final private static ThreadLocalSet session = new ThreadLocalSet(); /* * Construction * param serverAddress. Zookeeper servers address-port like :2181,:2181,. * param ownID. Source Key or lock name. * throws LockException */public MutexLockImplZooKeeper(String serverAddress, String ownID) throws LockException this(serverAddress, ownID, 0, 100, 10000);/* * Construction * param serverAddress. Zookeeper servers address-port like :2181,:2181,. * param ownID. Source Key or lock name. * param retry. Retry times. * param waitTime. Wait time. * param timeout. Time out. * throws LockException */public MutexLockImplZooKeeper(String serverAddress, String ownID, int retry, int waitTime, int timeout) throws LockExceptionthis.serverAddress = serverAddress;this.ownID = ownID;this.retry = retry;this.waitTime = waitTime;this.timeout = timeout;this.step = 1;this.count = 0;Overridesynchronized public void lock() throws LockException if(step!=1)throw new LockException(new Exception(This lock is invalid.);elsestep = 2;init();while(true & !isReentry)count+;tryexecute();break;catch(LockException ex)if(count=retry)closeLocalThread();closeZookeeper();throw new LockException(ex);elsetryThread.sleep(waitTime);catch(Exception e)private void init() throws LockExceptionif(session.get()=null | !session.get().contains(ownID)connectZooKeeper();Set set = session.get()=null ? new HashSet() : session.get();set.add(ownID);session.set(set);isReentry = false;elseisReentry = true;private void execute() throws LockExceptiontry zk.create(createPath(), new byte0, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); catch (KeeperException e) throw new LockException(e); catch (InterruptedException e) throw new LockException(e);private String createPath()tryif(zk.exists(root, false)=null)zk.create(root, new byte0, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); catch (KeeperException e) catch (InterruptedException e) return root+/+ownID;private void connectZooKeeper()throws LockException try if(zk=null) zk = new ZooKeeper(serverAddress, timeout, null); catch (IOException e) throw new LockException(e);private void closeLocalThread()if(!isReentry & session.get()!=null)if(session.get().contains(ownID) session.get().remove(ownID);private void closeZookeeper()if(zk!=null)try zk.close(); catch (InterruptedException e) finallyzk = null;Overridesynchronized public void close() throws LockException if(step!=2)throw new LockException(new Exception(This lock is invalid.);elsestep = 3;closeLocalThread();closeZookeeper(); DBLock.java/(c) Copyright Beyond PAAS contributors and others 2013package com.beyond.paas.tool.lock;/* * DBLock is interface. * author su xiaohui * CreateDate 2013-10-25 */public interface DBLock enum IsolateLevel READUNCOMMIT, READCOMMIT, NOREPEAT, SERIALIZE;/* * To get read lock * param group. It is group name, which is often organization name. * param table. It is table name. * throws