分布式鎖實(shí)現(xiàn)的正確打開(kāi)方式

public static void main(String[] args) throws Exception {
        //創(chuàng)建zookeeper的客戶(hù)端
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

        CuratorFramework client = CuratorFrameworkFactory.newClient("10.21.41.181:2181,10.21.42.47:2181,10.21.49.252:2181", retryPolicy);

        client.start();

        //創(chuàng)建分布式鎖, 鎖空間的根節(jié)點(diǎn)路徑為/curator/lock
        InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");

        mutex.acquire();

        //獲得了鎖, 進(jìn)行業(yè)務(wù)流程
        System.out.println("Enter mutex");

        //完成業(yè)務(wù)流程, 釋放鎖
        mutex.release();

        //關(guān)閉客戶(hù)端
        client.close();

    }

/**
    * Zookeeper客戶(hù)端
    */
private ZooKeeper zookeeper;

/**
    * 鎖的唯一標(biāo)識(shí)
    */
private String lockId;

/**
    * 與Zookeeper建立會(huì)話(huà)的信號(hào)量
    */
private CountDownLatch connectedLatch;

/**
    * 創(chuàng)建分布式鎖的過(guò)程中，開(kāi)始和等待請(qǐng)求創(chuàng)建分布式鎖的信號(hào)標(biāo)志
    */
private CountDownLatch creatingLatch;

/**
    * 分布式鎖路徑前綴
    */
private String locksRootPath = "/locks";

/**
    * 排在當(dāng)前節(jié)點(diǎn)前面一位的節(jié)點(diǎn)的路徑
    */
private String waitNodeLockPath;

/**
    * 為了獲得鎖，本次創(chuàng)建的節(jié)點(diǎn)的路徑
    */
private String currentNodeLockPath;

public ZookeeperTempOrderLock(String lockId) {
    this.lockId = lockId;
    try {
        // 會(huì)話(huà)超時(shí)時(shí)間
        int sessionTimeout = 30000;
        //
        zookeeper = new ZooKeeper("192.168.0.93:2181", sessionTimeout, this);
        connectedLatch.await();
    } catch (IOException ioe) {
        log.error("與Zookeeper建立連接時(shí)出現(xiàn)異常", ioe);
    } catch (InterruptedException ite) {
        log.error("等待與Zookeeper會(huì)話(huà)建立完成時(shí)出現(xiàn)異常", ite);
    }
}

@Override
public void process(WatchedEvent event) {
    if (Event.KeeperState.SyncConnected == event.getState()) {
        connectedLatch.countDown();
    }

    if (creatingLatch != null) {
        creatingLatch.countDown();
    }
}

/**
    * 獲取鎖
    */
public void acquireDistributedLock() {
    try {
        while(!tryLock()) {
            // 等待前一項(xiàng)服務(wù)釋放鎖的等待時(shí)間 不能超過(guò)一次Zookeeper會(huì)話(huà)的時(shí)間
            long waitForPreviousLockRelease = 30000;
            waitForLock(waitNodeLockPath, waitForPreviousLockRelease);
        }
    } catch (InterruptedException | KeeperException e) {
        log.error("等待上鎖的過(guò)程中出現(xiàn)異常", e);
    }
}

public boolean tryLock() {
    try {
        // 創(chuàng)建順序臨時(shí)節(jié)點(diǎn)
        currentNodeLockPath = zookeeper.create(locksRootPath + "/" + lockId,
                "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        // 查看剛剛創(chuàng)建的節(jié)點(diǎn)是不是最小節(jié)點(diǎn)
        // 比如針對(duì)于這個(gè)同名節(jié)點(diǎn)，之前有其它服務(wù)曾申請(qǐng)創(chuàng)建過(guò)，因此Zookeeper中臨時(shí)順序節(jié)點(diǎn)形如:
        // /locks/10000000000, /locks/10000000001, /locks/10000000002
        List<String> nodePaths = zookeeper.getChildren(locksRootPath, false);
        Collections.sort(nodePaths);
        if(currentNodeLockPath.equals(locksRootPath + "/" + nodePaths.get(0))) {
            // 如果是最小節(jié)點(diǎn)，則代表獲取到鎖
            return true;
        }
        // 如果不是最小節(jié)點(diǎn)，則找到比自己小1的節(jié)點(diǎn) (緊挨著自己)
        int previousLockNodeIndex = -1;
        for (int i = 0; i < nodePaths.size(); i++) {
            if(currentNodeLockPath.equals(locksRootPath + "/" + nodePaths.get(i))) {
                previousLockNodeIndex = i-1;
                break;
            }
        }
        this.waitNodeLockPath = nodePaths.get(previousLockNodeIndex);

    } catch (KeeperException | InterruptedException e) {
        log.error("創(chuàng)建臨時(shí)順序節(jié)點(diǎn)失敗", e);
    }
    return false;
}

/**
    * 等待其他服務(wù)釋放鎖
    * 實(shí)際上就是在等待前一個(gè)臨時(shí)節(jié)點(diǎn)被刪除
    *
    * @param nodePath 希望被刪除的節(jié)點(diǎn)的相對(duì)路徑
    * @param waitTime 等待時(shí)長(zhǎng) 單位:毫秒
    */
private boolean waitForLock(String nodePath, long waitTime) throws KeeperException, InterruptedException {
    Stat stat = zookeeper.exists(locksRootPath + "/" + nodePath, true);
    if (stat != null) {
        this.creatingLatch = new CountDownLatch(1);
        this.creatingLatch.await(waitTime, TimeUnit.MILLISECONDS);
        this.creatingLatch = null;
    }
    return true;
}

/**
    * 釋放鎖
    * 實(shí)際上就是刪除當(dāng)前創(chuàng)建的臨時(shí)節(jié)點(diǎn)
    */
public void releaseLock() {
    log.info("準(zhǔn)備刪除的節(jié)點(diǎn)路徑: " + currentNodeLockPath);
    try {
        zookeeper.delete(currentNodeLockPath, -1);
        currentNodeLockPath = null;
        zookeeper.close();
    } catch (Exception e) {
        log.error("刪除節(jié)點(diǎn)失敗", e);
    }
}

    /**
     * 與Zookeeper成功建立連接的信號(hào)標(biāo)志
     */
    private CountDownLatch connectedSemaphore = new CountDownLatch(1);

    /**
     * 創(chuàng)建分布式鎖的過(guò)程中，開(kāi)始和等待請(qǐng)求創(chuàng)建分布式鎖的信號(hào)標(biāo)志
     */
    private CountDownLatch creatingSemaphore;

    /**
     * Zookeeper客戶(hù)端
     */
    private ZooKeeper zookeeper;

    /**
     * 分布式鎖的過(guò)期時(shí)間 單位:毫秒
     */
    private static final Long DISTRIBUTED_KEY_OVERDUE_TIME = 30000L;

public ZookeeperLock() {
    try {
        this.zookeeper = new ZooKeeper("192.168.0.93:2181", 5000, new ZookeeperWatcher());
        try {
            connectedSemaphore.await();
        } catch (InterruptedException ite) {
            log.error("等待Zookeeper成功建立連接的過(guò)程中，線(xiàn)程拋出異常", ite);
        }
        log.info("與Zookeeper成功建立連接");
    } catch (Exception e) {
        log.error("與Zookeeper建立連接時(shí)出現(xiàn)異常", e);
    }
}

public boolean acquireDistributeLock(Long lockId) {
    String path = "/product-lock-" + lockId;

    try {
        zookeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        log.info("ThreadId=" + Thread.currentThread().getId() + "創(chuàng)建臨時(shí)節(jié)點(diǎn)成功");
        return true;
    } catch (Exception e) {
        // 若臨時(shí)節(jié)點(diǎn)已存在，則會(huì)拋出異常: NodeExistsException
        while (true) {
            // 相當(dāng)于給znode注冊(cè)了一個(gè)監(jiān)聽(tīng)器，查看監(jiān)聽(tīng)器是否存在
            try {
                Stat stat = zookeeper.exists(path, true);
                if (stat != null) {
                    this.creatingSemaphore = new CountDownLatch(1);
                    this.creatingSemaphore.await(DISTRIBUTED_KEY_OVERDUE_TIME, TimeUnit.MILLISECONDS);
                    this.creatingSemaphore = null;
                }
                zookeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                return true;
            } catch (Exception ex) {
                log.error("ThreadId=" + Thread.currentThread().getId() + ",查看臨時(shí)節(jié)點(diǎn)時(shí)出現(xiàn)異常", ex);
            }
        }
    }
}

public void releaseDistributedLock(Long lockId) {
    String path = "/product-lock-" + lockId;
    try {
        // 第二個(gè)參數(shù)version是數(shù)據(jù)版本 每次znode內(nèi)數(shù)據(jù)發(fā)生變化，都會(huì)使version自增，但由于分布式鎖創(chuàng)建的臨時(shí)znode沒(méi)有存數(shù)據(jù)，因此version=-1
        zookeeper.delete(path, -1);
        log.info("成功釋放分布式鎖, lockId=" + lockId + ", ThreadId=" + Thread.currentThread().getId());
    } catch (Exception e) {
        log.error("釋放分布式鎖失敗，lockId=" + lockId, e);
    }
}

private class ZookeeperWatcher implements Watcher {
    @Override
    public void process(WatchedEvent event) {
        log.info("接收到事件: " + event.getState() + ", ThreadId=" + Thread.currentThread().getId());

        if (Event.KeeperState.SyncConnected == event.getState()) {
            connectedSemaphore.countDown();
        }

        if (creatingSemaphore != null) {
            creatingSemaphore.countDown();
        }
    }
}

public static void main(String[] args) throws InterruptedException{
    long lockId = 20200730;

    new Thread(() ->{
        ZookeeperLock zookeeperLock = new ZookeeperLock();
        System.out.println("ThreadId1=" + Thread.currentThread().getId());
        System.out.println("ThreadId=" + Thread.currentThread().getId() + "獲取到分布式鎖: " + zookeeperLock.acquireDistributeLock(lockId));
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            log.error("ThreadId=" + Thread.currentThread().getId() + "暫停時(shí)出現(xiàn)異常", e);
        }
        zookeeperLock.releaseDistributedLock(lockId);
    }).start();

    TimeUnit.SECONDS.sleep(1);
    new Thread(() -> {
        ZookeeperLock zookeeperLock = new ZookeeperLock();
        System.out.println("ThreadId2=" + Thread.currentThread().getId());
        System.out.println("ThreadId=" + Thread.currentThread().getId() + "獲取到分布式鎖: " + zookeeperLock.acquireDistributeLock(lockId));
    }).start();
}

public String deductStock() {
    String lockKey = "product_001";
    try {
       /*Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "aaa"); //jedis.setnx
        stringRedisTemplate.expire(lockKey, 30, TimeUnit.SECONDS); //設(shè)置超時(shí)*/
        //為解決原子性問(wèn)題將設(shè)置鎖和設(shè)置超時(shí)時(shí)間合并
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "aaa", 10, TimeUnit.SECONDS);

        //未設(shè)置成功，當(dāng)前key已經(jīng)存在了，直接返回錯(cuò)誤
        if (!result) {
            return "error_code";
        }
        //業(yè)務(wù)邏輯實(shí)現(xiàn)，扣減庫(kù)存
        ....
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        stringRedisTemplate.delete(lockKey);
    }
    return "end";
}

RLock redissonLock = redission.getLock（）；
redissionLock.lock(30,TmieUnit.SECONDS)；加鎖并設(shè)置鎖的存活時(shí)間
redissionLock.unLock()；解鎖

public String deductStockRedission() {
    String lockKey = "product_001";
    RLock rlock = redission.getLock(lockKey);
    try {
        rlock.lock();

        //業(yè)務(wù)邏輯實(shí)現(xiàn)，扣減庫(kù)存
        ....
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        rlock.unlock();
    }
    return "end";
}

public String deductStockRedlock() {
    String lockKey = "product_001";
    //TODO 這里需要自己實(shí)例化不同redis實(shí)例的redission客戶(hù)端連接，這里只是偽代碼用一個(gè)redisson客戶(hù)端簡(jiǎn)化了
    RLock rLock1 = redisson.getLock(lockKey);
    RLock rLock2 = redisson.getLock(lockKey);
    RLock rLock3 = redisson.getLock(lockKey);

    // 向3個(gè)redis實(shí)例嘗試加鎖
    RedissonRedLock redLock = new RedissionRedLock(rLock1, rLock2, rLock3);
    boolean isLock;
    try {
        // 500ms拿不到鎖, 就認(rèn)為獲取鎖失敗。10000ms即10s是鎖失效時(shí)間。
        isLock = redLock.tryLock(500, 10000, TimeUnit.MILLISECONDS);
        System.out.println("isLock = " + isLock);
        if (isLock) {
            //業(yè)務(wù)邏輯處理
            ...
        }
    } catch (Exception e) {

    } finally {
        // 無(wú)論如何, 最后都要解鎖
        redLock.unlock();
    }
}