引

　　在分散式系統中，資源可能同時被多個客戶端申請訪問，因此保證資料訪問的正確性和效能是分散式系統必須要考慮的問題。非分散式下我們通常是通過synchronize或lock，以及資料庫鎖(不限制非分散式和分散式)，而這兩種多存在相應的弊端，synchronize或lock不能解決分散式系統，資料庫鎖在大量請求下容易產生鎖等待、死鎖和處理失敗對資料庫的影響較大。所以分散式鎖的應用成為大多數的首選。

Zookeeper分散式鎖

　　完全分散式鎖是全域性同步的，這意味著在任何時刻沒有兩個客戶端會同時認為它們都擁有相同的鎖，Zookeeper 可以實現分散式鎖。

　　Zookeeper分散式鎖的實現流程

　　Zookeeper實現分散式鎖是通過節點和臨時順序節點來實現的：

　　1.在建構函式裡面啟動的時候建立一個節點，假如命名為：lock。節點型別為持久節點（PERSISTENT）【ZooKeeper裡面的znode節點會自動同步的，而且是強一致性，建立一個節點後只有ZooKeeper叢集同步完成後算成功】

　　2.每當程序需要訪問共享資源時，會在lock節點下面建立響應的順序子節點，節點型別為臨時順序節點（EPHEMERAL_SEQUENTIAL）

　　3.在建立子節點之後，判斷剛剛建立的子節點順序號是否為最小節點，如果是最小節點，則可以獲得該鎖對資源進行訪問。（臨時子節點建立會自動生成一個序號的）

　　4.如果不是該節點，就獲得該節點的上一順序節點，並給該節點是否存在註冊監聽事件。同時在這裡阻塞。等待監聽事件的發生。獲得控制權（實現watch介面，並且重寫process方法，在process裡面實現監聽）

　　5.當完成之後，關閉ZooKeeper連線，進而可以應發監聽事件，釋放該鎖（客戶端關閉ZooKeeper連線之後會刪除當前的臨時節點）

具體實現

　　我們先用zookeeper包去實現這個流程，稍後在用Curator去實現（輪子）

/**
 * 
 * @author J
 */
public class DistributedLock implements Lock, Watcher {
    private ZooKeeper zk = null;
    // 根節點
    private String ROOT_LOCK = "/locks";
    // 競爭的資源
    private String lockName;
    // 等待的前一個鎖
    private String WAIT_LOCK;
    // 當前鎖
    private String CURRENT_LOCK;
    // 計數器
    private CountDownLatch countDownLatch;
    private int sessionTimeout = 30000;
    private List<Exception> exceptionList = new ArrayList<Exception>();

    /**
     * 配置分散式鎖
     * @param config 連線的url
     * @param lockName 競爭資源
     */
    public DistributedLock(String config, String lockName) {
        this.lockName = lockName;
        try {
            // 連線zookeeper
            zk = new ZooKeeper(config, sessionTimeout, this);
            Stat stat = zk.exists(ROOT_LOCK, false);
            if (stat == null) {
                // 如果根節點不存在，則建立根節點
                zk.create(ROOT_LOCK, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    /**
     * 節點監視器
     */
    public void process(WatchedEvent event) {
        if (this.countDownLatch != null) {
            this.countDownLatch.countDown();
        }
    }

    public void lock() {
        if (exceptionList.size() > 0) {
            throw new LockException(exceptionList.get(0));
        }
        try {
            if (this.tryLock()) {
                System.out.println(Thread.currentThread().getName() + " " + lockName + "獲得了鎖");
                return;
            } else {
                // 等待鎖
                waitForLock(WAIT_LOCK, sessionTimeout);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public boolean tryLock() {
        try {
            String splitStr = "_lock_";
            if (lockName.contains(splitStr)) {
                throw new LockException("鎖名有誤");
            }
            // 建立臨時有序節點
            CURRENT_LOCK = zk.create(ROOT_LOCK + "/" + lockName + splitStr, new byte[0],
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(CURRENT_LOCK + " 已經建立");
            // 取所有子節點
            List<String> subNodes = zk.getChildren(ROOT_LOCK, false);
            // 取出所有lockName的鎖
            List<String> lockObjects = new ArrayList<String>();
            for (String node : subNodes) {
                String _node = node.split(splitStr)[0];
                if (_node.equals(lockName)) {
                    lockObjects.add(node);
                }
            }
            Collections.sort(lockObjects);
            System.out.println(Thread.currentThread().getName() + " 的鎖是 " + CURRENT_LOCK);
            // 若當前節點為最小節點，則獲取鎖成功
            if (CURRENT_LOCK.equals(ROOT_LOCK + "/" + lockObjects.get(0))) {
                return true;
            }

            // 若不是最小節點，則找到自己的前一個節點
            String prevNode = CURRENT_LOCK.substring(CURRENT_LOCK.lastIndexOf("/") + 1);
            WAIT_LOCK = lockObjects.get(Collections.binarySearch(lockObjects, prevNode) - 1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean tryLock(long timeout, TimeUnit unit) {
        try {
            if (this.tryLock()) {
                return true;
            }
            return waitForLock(WAIT_LOCK, timeout);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    // 等待鎖
    private boolean waitForLock(String prev, long waitTime) throws KeeperException, InterruptedException {
        Stat stat = zk.exists(ROOT_LOCK + "/" + prev, true);

        if (stat != null) {
            System.out.println(Thread.currentThread().getName() + "等待鎖 " + ROOT_LOCK + "/" + prev);
            this.countDownLatch = new CountDownLatch(1);
            // 計數等待，若等到前一個節點消失，則precess中進行countDown，停止等待，獲取鎖
            this.countDownLatch.await(waitTime, TimeUnit.MILLISECONDS);
            this.countDownLatch = null;
            System.out.println(Thread.currentThread().getName() + " 等到了鎖");
        }
        return true;
    }

    public void unlock() {
        try {
            System.out.println("釋放鎖 " + CURRENT_LOCK);
            zk.delete(CURRENT_LOCK, -1);
            CURRENT_LOCK = null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public Condition newCondition() {
        return null;
    }

    public void lockInterruptibly() throws InterruptedException {
        this.lock();
    }


    public class LockException extends RuntimeException {
        private static final long serialVersionUID = 1L;
        public LockException(String e){
            super(e);
        }
        public LockException(Exception e){
            super(e);
        }
    }
}
 

public static void main(String[] args) {
		int n = 500;
        Runnable runnable = new Runnable() {
            public void run() {
                DistributedLock lock = null;
                try {
                    lock = new DistributedLock("127.0.0.1:2181", "lock");
                    lock.lock();
                    System.out.println(n);
                    System.out.println(Thread.currentThread().getName() + "正在執行");
                } finally {
                    if (lock != null) {
                        lock.unlock();
                    }
                }
            }
        };

        for (int i = 0; i < 10; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
    }

/**
 * 分散式鎖
 * @author J
 */
public class DistLock {

    private static final String rootPath = "distLock";

    private static final int DEFAULT_CONNECT_TIMEOUT = 2000;

    private static final int DEFAULT_SESSION_TIMEOUT = 2000;
    
    private static final int DEFAULT_LOCK_TIMEOUT = 120;

    private String connectionString;

    private int connectionTimeout;

    private int sessionTimeout;

    private CuratorFramework client;
    
    private Map<String, InterProcessMutex> locks;

    public DistLock() {
        this.connectionString = PropertyUtil.get("zkCluster");
        this.connectionTimeout =
                StringUtils.isBlank(PropertyUtil.get("connect.timeout")) ? DEFAULT_CONNECT_TIMEOUT
                        : Integer.valueOf(PropertyUtil.get("connect.timeout"));
        this.sessionTimeout =
                StringUtils.isBlank(PropertyUtil.get("request.timeout")) ? DEFAULT_SESSION_TIMEOUT
                        : Integer.valueOf(PropertyUtil.get("request.timeout"));
        this.client =
                CuratorFrameworkFactory.builder().connectString(connectionString)
                        .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                        .connectionTimeoutMs(this.connectionTimeout)
                        .sessionTimeoutMs(this.sessionTimeout).namespace(rootPath).build();
        this.client.start();
        locks = new HashMap<>(32);
    }

    /**
     * 
     * <p>Descrption: 獲取zk客戶端</p>
     * @Author J
     * @return CuratorFramework
     * @return
     */
    public CuratorFramework getClient() {
        return client;
    }

    /**
     * 
     * <p>Descrption: 獲取分散式鎖</p>
     * @Author J
     * @return boolean
     * @param action
     * @param lockId
     * @param time
     * @return
     * @throws Exception
     */
    public boolean lock(String action, String lockId, int time) throws Exception {
        String uniqueLockId = action+"_"+lockId;
        
        InterProcessMutex lock = new InterProcessMutex(this.client, "/"+uniqueLockId);
        boolean isLocked = lock.acquire(time, TimeUnit.SECONDS);
        if (isLocked) {
            this.locks.put(uniqueLockId, lock);
        }
        return isLocked;
    }
    
    /**
     * 
     * <p>Descrption: 獲取分散式鎖</p>
     * @Author J
     * @return boolean
     * @param action
     * @param lockId
     * @return
     * @throws Exception
     */
    public boolean lock(String action, String lockId) throws Exception {
        return lock(action, lockId, DEFAULT_LOCK_TIMEOUT);
    }
    
    /**
     * 
     * <p>Descrption: 批量鎖</p>
     * @Author J
     * @return boolean
     * @param action
     * @param lockIds
     * @param time
     * @return
     * @throws Exception
     */
    
    public boolean batchLock(String action, Collection<String> lockIds, int time) throws Exception {
        boolean ret = true;
        for (String lockId : lockIds) {
            ret = ret && lock(action, lockId, time);
            // 所有加鎖成功才算成功
            if (!ret) {
                unBatchLock(action, lockIds);
                break;
            }
        }
        return ret;
    }
    
    public boolean batchLock(String action, Collection<String> lockIds) throws Exception {
        return batchLock(action, lockIds, DEFAULT_LOCK_TIMEOUT);
    }
    
    /**
     * 
     * <p>Descrption: 返回無法獲取鎖的lockId</p>
     * @Author J
     * @return Set<String>
     * @param action
     * @param lockIds
     * @param time
     * @return
     * @throws Exception
     */
    public Set<String> batchLockRetUnlock(String action, Collection<String> lockIds, int time) throws Exception {
        Set<String> unlockIds = new HashSet<>();
        for (String lockId : lockIds) {
            if (!lock(action, lockId, time)){
                unlockIds.add(lockId);
            }
        }
        return unlockIds;
    }
    
    public Set<String> batchLockRetUnlock(String action, Collection<String> lockIds) throws Exception {
        return batchLockRetUnlock(action, lockIds, DEFAULT_LOCK_TIMEOUT);
    }
    
    /**
     * 
     * <p>Descrption: 併發批量加鎖</p>
     * @Author J
     * @return Set<String>
     * @param action
     * @param lockIds
     * @param time
     * @return
     * @throws Exception
     */
    public Set<String> batchParalleLockRetUnlock(String action, Collection<String> lockIds, int time) throws Exception {
        final Set<String> unlockIds = new HashSet<>();
//        lockIds.parallelStream().limit(100).forEach(lockId -> {
    	for(String lockId : lockIds){
            try {
                if (!lock(action, lockId, time)) {
                    unlockIds.add(lockId);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        };
        return unlockIds;
    }
    
    public Set<String> batchParalleLockRetUnlock(String action, Collection<String> lockIds) throws Exception {
        return batchParalleLockRetUnlock(action, lockIds, DEFAULT_LOCK_TIMEOUT);
    }
    
    /**
     * 
     * <p>Descrption: 釋放鎖</p>
     * @Author J
     * @return void
     * @param action
     * @param lockId
     * @throws Exception
     */
    public void unlock(String action, String lockId) throws Exception {
        String uniqueLockId = action+"_"+lockId;
        InterProcessMutex lock = null;
        if ((lock = this.locks.get(uniqueLockId)) != null) {
            this.locks.remove(uniqueLockId);
            lock.release();
        }
    }
    
    /**
     * 
     * <p>Descrption: 批量釋放鎖</p>
     * @Author J
     * @return void
     * @param action
     * @param lockId
     * @throws Exception 
     */
    public void unBatchLock(String action, Collection<String> lockIds) {
        for (String lockId : lockIds) {
            try {
                unlock(action, lockId);
            } catch (Exception e) {
                continue;
            }
        }
    }
    
    /**
     * 
     * <p>Descrption: 併發批量解鎖</p>
     * @Author J
     * @return void
     * @param action
     * @param lockIds
     */
    public void unBatchParalleLock(String action, Collection<String> lockIds) {
//        lockIds.parallelStream().limit(100).forEach(lockId -> {
    	for(String lockId : lockIds){
            try {
                unlock(action, lockId);
            } catch (Exception e) {
                // DO NOTHING
            }
        };
    }
    
    /**
     * 
     * <p>Descrption: 關閉客戶端</p>
     * @Author J
     * @return void
     */
    public void close() {
        if (this.client != null) {
            CloseableUtils.closeQuietly(this.client);
        }
    }
}

    @Autowired
    private DistLock distLock;
	
	@Test
	public void lockTest() throws Exception {
		
		try {
			if(distLock.lock("testlock", "BatchInsertTest", 10)){
				System.out.println(1);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			distLock.unlock("kft", "BatchDebitKFT");
		}
	}