摘要：分布式系統(tǒng)錯(cuò)綜復(fù)雜，今天，我們著重對(duì)分布式系統(tǒng)的互斥性與冪等性進(jìn)行分析與解決。阻塞鎖與自旋鎖。公平鎖與非公平鎖。實(shí)現(xiàn)今天重點(diǎn)講解使用實(shí)現(xiàn)分布式鎖。個(gè)人感覺是最適合實(shí)現(xiàn)分布式鎖。如以上流程，接口無法冪等，可能導(dǎo)致重復(fù)扣款。

隨著數(shù)據(jù)量的增大，用戶的增多，系統(tǒng)的并發(fā)訪問越來越大，傳統(tǒng)的單機(jī)已經(jīng)滿足不了需求，分布式系統(tǒng)成為一種必然的趨勢(shì)。分布式系統(tǒng)錯(cuò)綜復(fù)雜，今天，我們著重對(duì)分布式系統(tǒng)的互斥性與冪等性進(jìn)行分析與解決。

互斥性問題也就是共享資源的搶占問題。如何解決呢？也就是鎖，保證對(duì)共享資源的串行化訪問?；コ庑砸绾螌?shí)現(xiàn)？。在java中，最常用的是synchronized和lock這兩種內(nèi)置的鎖，但這只適用于單進(jìn)程中的多線程。對(duì)于在同一操作系統(tǒng)下的多個(gè)進(jìn)程間，常見的鎖實(shí)現(xiàn)有pv信號(hào)量等。然而，當(dāng)問題擴(kuò)展到多臺(tái)機(jī)器的多個(gè)操作系統(tǒng)時(shí)，也就是分布式鎖，情況就復(fù)雜多了。

鎖要存在哪里。必須提供一個(gè)所有主機(jī)都能訪問到的存儲(chǔ)空間

加鎖的進(jìn)程在掛掉之后，如何確保鎖被解開，釋放資源。可以通過超時(shí)機(jī)制或者定時(shí)檢測(cè)心跳來實(shí)現(xiàn)

不同進(jìn)程間如何獲取相同的唯一標(biāo)識(shí)來競(jìng)爭(zhēng)鎖。可以利用要保護(hù)的資源生成一個(gè)唯一的id

獲取鎖操作的原子性。必須保證讀取鎖狀態(tài)、加鎖兩步的原子性

鎖的可重入性。某個(gè)線程試圖再次獲取由自己持有的鎖，這個(gè)操作會(huì)百分百成功，這就是可重入性。如果不能保證可重入性，就會(huì)有死鎖的可能。

阻塞鎖與自旋鎖。當(dāng)獲取不到鎖時(shí)，阻塞鎖就是線程阻塞自身，等待喚醒，自旋鎖就是不斷的嘗試重新獲取鎖。

公平鎖與非公平鎖。公平鎖保證按照請(qǐng)求的順序獲取鎖，非公平鎖就是可以插隊(duì)。公平鎖一般要維持一個(gè)隊(duì)列來實(shí)現(xiàn)，所以非公平鎖的性能會(huì)更好一點(diǎn)。

避免驚群效應(yīng)。如果分布式鎖是阻塞鎖，當(dāng)鎖的占有者釋放鎖時(shí)，要避免同時(shí)喚醒多個(gè)阻塞的線程，產(chǎn)生驚群效應(yīng)。

今天重點(diǎn)講解使用zookeeper實(shí)現(xiàn)分布式鎖。個(gè)人感覺zookeeper是最適合實(shí)現(xiàn)分布式鎖。它的幾個(gè)特性：

順序節(jié)點(diǎn)：可以避免驚群效應(yīng)

臨時(shí)節(jié)點(diǎn)：避免機(jī)器宕機(jī)倒是鎖無法釋放

watch機(jī)制：可以及時(shí)喚醒等待的線程

zk實(shí)現(xiàn)分布式鎖的流程如下

我這里用zk實(shí)現(xiàn)了一個(gè)可重入的、阻塞的、公平的分布式鎖，代碼如下：

package locks;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import utils.ZkUtils;
import watcher.PredecessorNodeWatcher;
import watcher.SessionWatcher;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
* Created by huangwt on 2018/3/21.
*/
@Slf4j
public class ReentrantZKLock {

    private final static String BASE_NODE = "/baseNode";
    private final static String CHILDREN_NODE = "/node_";

    private final Lock localLock;
    private final Condition condition;

    //用于重入檢測(cè)
    private static ThreadLocal threadLocal = new ThreadLocal();

    private ZooKeeper zooKeeper = null;

    private String node = null;

    ReentrantZKLock(String addr, int timeout) {
        try {
            zooKeeper = new ZooKeeper(addr, timeout, new SessionWatcher());
            localLock = new ReentrantLock();
            condition = localLock.newCondition();
        } catch (IOException e) {
            log.error("get zookeeper failed", e);
            throw new RuntimeException(e);
        }
    }

    public void lock() {
        //重入檢測(cè)
        if (checkReentrant()) {
            return;
        }
        try {
            node = zooKeeper.create(BASE_NODE + CHILDREN_NODE, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            while (true) {
                localLock.lock();
                try {
                    List childrenNodes = zooKeeper.getChildren(BASE_NODE, false);
                    ZkUtils.childNodeSort(childrenNodes);
                    //當(dāng)前節(jié)點(diǎn)的索引
                    int myNodeIndex = childrenNodes.indexOf(node);
                    //當(dāng)前節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn)
                    int beforeNodeIndex = myNodeIndex - 1;
                    Stat stat = null;
                    while (beforeNodeIndex >= 0) {
                        stat = zooKeeper.exists(childrenNodes.get(beforeNodeIndex), new PredecessorNodeWatcher(condition));
                        if (stat != null) {
                            break;
                        }
                    }

                    if (stat != null) {  //前序節(jié)點(diǎn)存在，等待前序節(jié)點(diǎn)被刪除，釋放鎖
                        condition.await();
                    } else { // 獲取到鎖
                        threadLocal.set(new AtomicInteger(1));
                        return;
                    }
                } finally {
                    localLock.unlock();
                }
            }
        } catch (Exception e) {
            log.error("lock failed", e);
            throw new RuntimeException(e);
        }

    }

    public void unlock() {
        AtomicInteger times = threadLocal.get();
        if (times == null) {
            return;
        }
        if (times.decrementAndGet() == 0) {
            threadLocal.remove();
            try {
                zooKeeper.delete(node, -1);
            } catch (Exception e) {
                log.error("unlock faild", e);
                throw new RuntimeException(e);
            }
        }

    }

    private boolean checkReentrant() {
        AtomicInteger times = threadLocal.get();
        if (times != null) {
            times.incrementAndGet();
            return true;
        }

        return false;
    }
}

package utils;

import java.util.Collections;
import java.util.Comparator;
import java.util.List;

/**
* Created by huangwt on 2018/3/24.
*/
public class ZkUtils {

    /**
    * 對(duì)子節(jié)點(diǎn)排序
    *
    * @param node
    */
    public static void childNodeSort(List node) {
        Collections.sort(node, new ChildNodeCompare());
    }

    private static class ChildNodeCompare implements Comparator {

        public int compare(String childNode1, String childNode2) {
            return childNode1.compareTo(childNode2);
        }
    }

}

package watcher;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

import java.util.concurrent.locks.Condition;

/**
* Created by huangwt on 2018/3/24.
*/
public class PredecessorNodeWatcher implements Watcher {
    private Condition condition = null;

    public PredecessorNodeWatcher(Condition condition) {
        this.condition = condition;
    }

    public void process(WatchedEvent event) {
        //前序節(jié)點(diǎn)被刪除，鎖被釋放，喚醒當(dāng)前等待線程
        if(event.getType() == Event.EventType.NodeDeleted){
            condition.signal();
        }
    }
}

package watcher;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

/**
* Created by huangwt on 2018/3/24.
*/
@Slf4j
public class SessionWatcher implements Watcher {
    public void process(WatchedEvent event) {
        if (event.getState() == Event.KeeperState.SyncConnected) {
            log.info("get zookeeper success");
        }
    }
}

主要是使用了ThreadLocal實(shí)現(xiàn)了鎖的可重入性，使用watch機(jī)制實(shí)現(xiàn)了阻塞鎖，使用臨時(shí)節(jié)點(diǎn)實(shí)現(xiàn)的公平鎖。
這段代碼只是一個(gè)demo供大家參考，還有很多問題沒解決。比如當(dāng)zookper掛掉的時(shí)候，阻塞的線程就無法被喚醒，這時(shí)候就需要監(jiān)聽zk的心跳。

冪等性是系統(tǒng)接口對(duì)外的一種承諾，數(shù)學(xué)表達(dá)為：f(f(x)) = f(x)。
冪等性指的是，使用相同參數(shù)對(duì)同一資源重復(fù)調(diào)用某個(gè)接口的結(jié)果與調(diào)用一次的結(jié)果相同。

假設(shè)現(xiàn)在有一個(gè)方法 ：Boolean withdraw(account_id, amount) ，作用是從account_id對(duì)應(yīng)的賬戶中扣除amount數(shù)額的錢，如果扣除成功則返回true，賬戶余額減少amount； 如果扣除失敗則返回false，賬戶余額不變。
如以上流程，接口無法冪等，可能導(dǎo)致重復(fù)扣款。

請(qǐng)求獲取ticketId

請(qǐng)求扣款，傳入ticketId

根據(jù)ticketId查詢此次操作是否存在，如果存在則表示該操作已經(jīng)執(zhí)行過，直接返回結(jié)果；如果不存在，扣款，保存結(jié)果

返回結(jié)果到客戶端