分布式锁（zookeeper）与接口幂等性实现

摘要：分布式系统错综复杂，今天，我们着重对分布式系统的互斥性与幂等性进行分析与解决。阻塞锁与自旋锁。公平锁与非公平锁。实现今天重点讲解使用实现分布式锁。个人感觉是最适合实现分布式锁。如以上流程，接口无法幂等，可能导致重复扣款。

随着数据量的增大，用户的增多，系统的并发访问越来越大，传统的单机已经满足不了需求，分布式系统成为一种必然的趋势。分布式系统错综复杂，今天，我们着重对分布式系统的互斥性与幂等性进行分析与解决。

互斥性问题也就是共享资源的抢占问题。如何解决呢？也就是锁，保证对共享资源的串行化访问。互斥性要如何实现？。在java中，最常用的是synchronized和lock这两种内置的锁，但这只适用于单进程中的多线程。对于在同一操作系统下的多个进程间，常见的锁实现有pv信号量等。然而，当问题扩展到多台机器的多个操作系统时，也就是分布式锁，情况就复杂多了。

锁要存在哪里。必须提供一个所有主机都能访问到的存储空间

加锁的进程在挂掉之后，如何确保锁被解开，释放资源。可以通过超时机制或者定时检测心跳来实现

不同进程间如何获取相同的唯一标识来竞争锁。可以利用要保护的资源生成一个唯一的id

获取锁操作的原子性。必须保证读取锁状态、加锁两步的原子性

锁的可重入性。某个线程试图再次获取由自己持有的锁，这个操作会百分百成功，这就是可重入性。如果不能保证可重入性，就会有死锁的可能。

阻塞锁与自旋锁。当获取不到锁时，阻塞锁就是线程阻塞自身，等待唤醒，自旋锁就是不断的尝试重新获取锁。

公平锁与非公平锁。公平锁保证按照请求的顺序获取锁，非公平锁就是可以插队。公平锁一般要维持一个队列来实现，所以非公平锁的性能会更好一点。

避免惊群效应。如果分布式锁是阻塞锁，当锁的占有者释放锁时，要避免同时唤醒多个阻塞的线程，产生惊群效应。

今天重点讲解使用zookeeper实现分布式锁。个人感觉zookeeper是最适合实现分布式锁。它的几个特性：

顺序节点：可以避免惊群效应

临时节点：避免机器宕机倒是锁无法释放

watch机制：可以及时唤醒等待的线程

zk实现分布式锁的流程如下

我这里用zk实现了一个可重入的、阻塞的、公平的分布式锁，代码如下：

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;

    //用于重入检测
    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() {
        //重入检测
        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);
                    //当前节点的索引
                    int myNodeIndex = childrenNodes.indexOf(node);
                    //当前节点的前一个节点
                    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) {  //前序节点存在，等待前序节点被删除，释放锁
                        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 {

    /**
    * 对子节点排序
    *
    * @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) {
        //前序节点被删除，锁被释放，唤醒当前等待线程
        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实现了锁的可重入性，使用watch机制实现了阻塞锁，使用临时节点实现的公平锁。
这段代码只是一个demo供大家参考，还有很多问题没解决。比如当zookper挂掉的时候，阻塞的线程就无法被唤醒，这时候就需要监听zk的心跳。

幂等性是系统接口对外的一种承诺，数学表达为：f(f(x)) = f(x)。
幂等性指的是，使用相同参数对同一资源重复调用某个接口的结果与调用一次的结果相同。

假设现在有一个方法 ：Boolean withdraw(account_id, amount) ，作用是从account_id对应的账户中扣除amount数额的钱，如果扣除成功则返回true，账户余额减少amount； 如果扣除失败则返回false，账户余额不变。
如以上流程，接口无法幂等，可能导致重复扣款。

请求获取ticketId

请求扣款，传入ticketId

根据ticketId查询此次操作是否存在，如果存在则表示该操作已经执行过，直接返回结果；如果不存在，扣款，保存结果

返回结果到客户端