Zookeeper實(shí)現(xiàn)分布式鎖總結(jié)

優(yōu)點(diǎn)：

• 有效的解決單點(diǎn)問(wèn)題，不可重入問(wèn)題，非阻塞問(wèn)題以及鎖無(wú)法釋放的問(wèn)題
• 實(shí)現(xiàn)較為簡(jiǎn)單

缺點(diǎn)：

• 性能上不如使用緩存實(shí)現(xiàn)的分布式鎖，因?yàn)槊看卧趧?chuàng)建鎖和釋放鎖的過(guò)程中，都要?jiǎng)討B(tài)創(chuàng)建、銷(xiāo)毀臨時(shí)節(jié)點(diǎn)來(lái)實(shí)現(xiàn)鎖功能
• 需要對(duì)Zookeeper的原理有所了解

排他鎖

排他鎖，又稱(chēng)寫(xiě)鎖或獨(dú)占鎖。如果事務(wù)T1對(duì)數(shù)據(jù)對(duì)象O1加上了排他鎖，那么在整個(gè)加鎖期間，只允許事務(wù)T1對(duì)O1進(jìn)行讀取或更新操作，其他任務(wù)事務(wù)都不能對(duì)這個(gè)數(shù)據(jù)對(duì)象進(jìn)行任何操作，直到T1釋放了排他鎖。

排他鎖核心是保證當(dāng)前有且僅有一個(gè)事務(wù)獲得鎖，并且鎖釋放之后，所有正在等待獲取鎖的事務(wù)都能夠被通知到。

Zookeeper 的強(qiáng)一致性特性，能夠很好地保證在分布式高并發(fā)情況下節(jié)點(diǎn)的創(chuàng)建一定能夠保證全局唯一性，即Zookeeper將會(huì)保證客戶(hù)端無(wú)法重復(fù)創(chuàng)建一個(gè)已經(jīng)存在的數(shù)據(jù)節(jié)點(diǎn)?？梢岳肸ookeeper這個(gè)特性，實(shí)現(xiàn)排他鎖。

• 定義鎖：通過(guò)Zookeeper上的數(shù)據(jù)節(jié)點(diǎn)來(lái)表示一個(gè)鎖
• 獲取鎖：客戶(hù)端通過(guò)調(diào)用 create 方法創(chuàng)建表示鎖的臨時(shí)節(jié)點(diǎn)，可以認(rèn)為創(chuàng)建成功的客戶(hù)端獲得了鎖，同時(shí)可以讓沒(méi)有獲得鎖的節(jié)點(diǎn)在該節(jié)點(diǎn)上注冊(cè)Watcher監(jiān)聽(tīng)，以便實(shí)時(shí)監(jiān)聽(tīng)到lock節(jié)點(diǎn)的變更情況
• 釋放鎖：以下兩種情況都可以讓鎖釋放
  • 當(dāng)前獲得鎖的客戶(hù)端發(fā)生宕機(jī)或異常，那么Zookeeper上這個(gè)臨時(shí)節(jié)點(diǎn)就會(huì)被刪除
  • 正常執(zhí)行完業(yè)務(wù)邏輯，客戶(hù)端主動(dòng)刪除自己創(chuàng)建的臨時(shí)節(jié)點(diǎn)

基于Zookeeper實(shí)現(xiàn)排他鎖流程：

img

代碼實(shí)現(xiàn)如下：

package zk.curator;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;

/**
 * Created by Joe on 2019/3/6.
 */
public class ExclusiveLock {
    private static final Logger logger = LoggerFactory.getLogger(ExclusiveLock.class);

    private static final String LOCK_NAMESPACE = "lock_space";
    private static final String LOCK_NODE = "exclusive_lock";
    // 用于掛起請(qǐng)求，并等待上一個(gè)分布式鎖的釋放
    private static CountDownLatch countDownLatch;
    private CuratorFramework client;
    private String name;

    public ExclusiveLock(String zkPath, String name) {
        this.name = name;
        // zk 客戶(hù)端初始化
        RetryPolicy retryPolicy = new RetryNTimes(3, 5000);
        client = CuratorFrameworkFactory.builder()
                .connectString(zkPath)
                .sessionTimeoutMs(10000)
                .retryPolicy(retryPolicy)
                .namespace(LOCK_NAMESPACE)
                .build();
        client.start();
        // countDownLatch 初始化
        countDownLatch = new CountDownLatch(1);
        init();
    }

    private void init() {
        try {
            if (Objects.isNull(client.checkExists().forPath("/"))) {
                client.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath("/");
            }
            addWatcherToLock("/");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void getLock() {
        // 使用死循環(huán)用于獲取請(qǐng)求
        while (true) {
            try {
                client.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath("/" + LOCK_NODE);
                logger.info("{} 分布式鎖獲取成功", name);
                return;
            } catch (Exception e) {
                e.printStackTrace();
                logger.info("{} 分布式鎖獲取失敗", name);
                try {
                    synchronized (ExclusiveLock.class) {
                        if (countDownLatch.getCount() <= 0) {
                            countDownLatch = new CountDownLatch(1);
                        }
                    }
                    countDownLatch.await();
                } catch (Exception e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    public boolean releaseLock() {
        try {
            if (Objects.nonNull(client.checkExists().forPath("/" + LOCK_NODE))) {
                client.delete().forPath("/" + LOCK_NODE);
            }
            logger.info("{} 分布式鎖釋放成功", name);
        } catch (Exception e) {
            logger.info("{} 分布式鎖釋放失敗", name);
            return false;
        }
        return true;
    }

    private void addWatcherToLock(String path) throws Exception {
        PathChildrenCache cache = new PathChildrenCache(client, path, true);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener(new ExclusiveLockWatcherListener());
    }

    private static class ExclusiveLockWatcherListener implements PathChildrenCacheListener {
        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
            // 監(jiān)聽(tīng)子節(jié)點(diǎn)被移除事件
            if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(event.getType())) {
                String path = event.getData().getPath();
                logger.info("上一個(gè)會(huì)話(huà)已經(jīng)釋放鎖 or 會(huì)話(huà)已經(jīng)端蓋，節(jié)點(diǎn)路徑為 {}", path);
                if (path.contains(LOCK_NODE)) {
                    System.out.println("釋放計(jì)數(shù)器，讓其他請(qǐng)求來(lái)獲得分布式鎖");
                    countDownLatch.countDown();
                }
            }
        }
    }
}

測(cè)試代碼：

package zk.curator;

/**
 * Created by Joe on 2019/3/6.
 */
public class Test {
    private static final String zkServerPath = "ip:port";

    public static void main(String[] args) {
        Thread[] threads = new Thread[5];
        for (int i = 0; i < 5; i++) {
            final int index = i;
            threads[i] = new Thread(() -> {
                ExclusiveLock exclusiveLock = new ExclusiveLock(zkServerPath, "thread" + index);
                exclusiveLock.getLock();
                exclusiveLock.releaseLock();
            });
            threads[i].start();
        }
    }
}