Java分布式锁由浅入深介绍

什么是分布式锁

分布式锁是一种通过共享锁来保证分布式环境下多进程、多线程之间数据同步的技术。常用的锁算法有互斥锁、读写锁、乐观锁、悲观锁等。

基于Zookeeper的分布式锁

Zookeeper是一种分布式协同管理工具，提供了一种基于节点的会话机制，这种机制可以通过锁节点来控制多个进程的协调。Zookeeper主要有以下特点：

• 一致性：所有客户端在同一时间看到同样的节点视图
• 分区容错性：容忍网络分区故障
• 原子性：一次操作要么成功要么失败

以下是基于Zookeeper实现的一个简单分布式锁的示例

public class ZookeeperLock implements Lock{
    private ZooKeeper zooKeeper;
    private String lockPath;
    private String lockName;
    private String currentNodeName;
    private ThreadLocal<AtomicInteger> reentryCount = new ThreadLocal<>();
    private static final String SEPARATOR = "/";

    public ZookeeperLock(String connectString, int sessionTimeout, String lockPath, String lockName) {
        this.zooKeeper = new ZooKeeper(connectString, sessionTimeout, event -> {});
        this.lockPath = SEPARATOR + lockPath;
        this.lockName = SEPARATOR + lockName;
        // 如果lockPath不存在就创建一个
        try{
            if(zooKeeper.exists(this.lockPath, false) == null){
                zooKeeper.create(this.lockPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void lock() {
        if(currentNodeName != null){
            // 如果已经拥有锁，则重入次数+1，返回
            reentryCount.get().incrementAndGet();
            return;
        }
        try{
            // 创建一个临时节点
            String nodeName = zooKeeper.create(lockPath + lockName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            currentNodeName = nodeName.substring(lockPath.length());
            // 如果获取锁失败，注册监听器
            while(true){
                List<String> children = zooKeeper.getChildren(lockPath, false);
                SortedSet<String> sortedNodeNames = new TreeSet<>();
                for(String child : children){
                    sortedNodeNames.add(lockPath + child);
                }
                String firstNodeName = sortedNodeNames.first();
                if(lockPath + currentNodeName.equals(firstNodeName)){
                    return;
                }
                CountDownLatch countDownLatch = new CountDownLatch(1);
                String previousNodeName = null;
                for(String nodeName : sortedNodeNames){
                    if((lockPath + currentNodeName).equals(nodeName)){
                        break;
                    }
                    previousNodeName = nodeName.substring(lockPath.length());
                }
                if(previousNodeName != null){
                    echoNodeCreatedWatcher(previousNodeName, countDownLatch);
                    countDownLatch.await();
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 注册监听器
     */
    private void echoNodeCreatedWatcher(String nodeName, CountDownLatch countDownLatch){
        try{
            zooKeeper.exists(lockPath + nodeName, event -> {
                if(event.getType() == NodeDeleted){
                    countDownLatch.countDown();
                }
            });
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void unlock() {
        if(currentNodeName == null){
            throw new RuntimeException("锁还未被获取");
        }
        int count = reentryCount.get().decrementAndGet();
        if(count > 0){
            return;
        }
        try{
            zooKeeper.delete(lockPath + currentNodeName, -1);
            currentNodeName = null;
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    ...
}

基于Redis的分布式锁

Redis也是一种分布式缓存中间件，提供了一种可重入的分布式锁，实现比Zookeeper简单，且性能也更高。

以下是基于Redis实现的一个简单分布式锁的示例

public class RedisLock implements Lock{
    private static final String LOCK_PREFIX = "redis_lock:";

    private Jedis jedis;
    private String lockKey;
    private int lockExpireMills;
    private String lockValue;
    private ThreadLocal<Integer> reentryCount;

    public RedisLock(Jedis jedis, String lockKey, int lockExpireMills){
        this.jedis = jedis;
        this.lockKey = LOCK_PREFIX + lockKey;
        this.lockExpireMills = lockExpireMills;
        this.reentryCount = ThreadLocal.withInitial(() -> 0);
    }

    @Override
    public void lock() {
        if(reentryCount.get() > 0){
            reentryCount.set(reentryCount.get() + 1);
            return;
        }
        lockValue = UUID.randomUUID().toString();
        String result = jedis.set(lockKey, lockValue, "NX", "PX", lockExpireMills);
        if(!"OK".equals(result)){
            throw new IllegalStateException("获取锁失败");
        }else{
            reentryCount.set(1);
        }
    }

    ...