吐血推荐-详解分布式锁(下)

「这是我参与11月更文挑战的第7天，活动详情查看：2021最后一次更文挑战」

MySql 实现分布式锁

使用 Mysql 实现分布式锁在实际开发中的应用场景比较少，一般只有在性能要求不是很高，不想要引入别的组件的时候才会使用。它最大的特点就是理解起来比较容易。

它主要有以下三种实现方式：

1. 基于表记录实现
2. 借助 mysql 的悲观锁实现

基于表记录实现

先创建一张表：

sql复制代码CREATE TABLE `mysql_lock` (
	`id` BIGINT NOT NULL AUTO_INCREMENT,
	`resource` int NOT NULL COMMENT '锁定的资源',
	`description` varchar(1024) NOT NULL DEFAULT "" COMMENT '描述',
	PRIMARY KEY (`id`),
	UNIQUE KEY `uiq_idx_resource` (`resource`) 
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='基于表记录实现';

申请锁操作就是在表中插入一条对应的记录：

sql复制代码INSERT INTO mysql_lock (resource, description) VALUES (1, '申请资源:1');

释放锁操作就是删除插入的那一条表记录即可：

sql复制代码DELETE FROM mysql_lock WHERE resource = 1;

实现原理说明：

我们在创建表的时候给 resource 加了唯一约束，它是不能重复插入的，这样就实现了互斥性。前面我们讲分布式锁的时候详细说明了分布式锁应该拥有的特性，上面的例子没有实现其他的特性，下面我们来说一下具体的优化方案：

1. 对于超时时间：可以写一个定时清理过期资源的程序
2. 对于可重入性，独占性：可以添加一个字段来记录线程的编号，如果是同一线程允许再次获取锁，每次删除数据库记录的时候校验线程编号，保证独占性，自己的锁只能自己解开。
3. 对于 mysql 的可靠性：可以设置主备或者集群来防止单点故障。
4. 这边还有一个小问题，就是每次去获取锁的时候，线程不是阻塞的只去插入向库里面插入一次，失败了也不重试，这个都需要在代码逻辑中自己实现了。

借助 mysql 悲观锁实现

为了提高分布式锁的效率，可以使用查询语句，借助 for update 关键字来给被查询的记录添加行锁中悲观锁，这样别的线程就没有办法对这条记录进行任何操作，从而达到保护共享资源的目的。

使用行锁需要注意的点

1. mysql 默认是会自动提交事务的，应该手动禁止一下：SET AUTOCOMMIT = 0;
2. 行锁是建立在索引的基础上的，如果查询时候不是走的索引的话，行锁会升级为表锁进行全表扫描。

我们继续使用上面那张表来进行说明：

1. 申请锁操作：SELECT * FROM mysql_lock WHERE id = 1 FOR UPDATE; 只要可以查的出来就是申请成功的，没有获取到的会被阻塞，阻塞的超时时可以通过设置 mysql 的 innodb_lock_wait_timeout 来进行设置。注意 WHERE id = 1 这个查询条件是走索引的。
2. 释放锁操作：COMMIT; 事务提交之后别的线程就可以查的这条记录了。

说明：这边简单提供了一下 mysql 实现分布式锁的两种思路，实际开发不建议使用，要使用的话建议使用第二种，像上面说的分布式锁应该实现的特性使用数据库的话，好多需要开发者手动去实现，不太友好。

Zookeeper 实现分布式锁

在使用 zookeeper 实现分布式锁之前我们先来了解一点前置知识：

zk 的节点类型

1. 持久化节点：客户端断开连接，节点还在
2. 持久化顺序节点：在持久化节点的基础上保证有序性
3. 临时节点：客户端断开连接，节点就删除了
4. 临时顺序节点：在临时节点的基础上保证有序性

实现 zk 分布式锁的思路

1. 利用 zk 同级节点的唯一特性可以实现锁的互斥性
2. 利用 zk 临时节点的特性可以避免正在占用锁的线程没释放锁就因为一些不可抗力因素宕机导致其他线程无法获取锁最终死锁的问题。
3. 利用 zk 的 节点的watcher 事件可以轻松实现通知其他线程争抢锁的功能。
4. 利用 zk 顺序节点的特性，可以实现公平锁，按照申请锁的顺序来唤醒阻塞的线程，防止羊群效应。

羊群效应：当并发量巨大的时候，只有一个线程会获得锁，很多线程就会阻塞，当获得锁的那个线程释放锁之后，其他所有在等待的线程就会一起争抢锁，可能会因为瞬间启动的线程过多而导致服务器挂到的情况，这就是所谓的羊群效应。

自定义 zk 分布式锁

代码示例：

java复制代码package com.aha.lock.zk;

import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.util.StringUtils;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * 使用 zk 实现分布式锁
 * 实现 Lock 遵循 JUC 提供的规范
 *
 * @author: WT
 * @date: 2021/11/23 16:41
 */
@Slf4j
public class ZkDistributedLock implements Lock {

    /**
     * 用于协调线程的执行时机，注意：countdown 之后，再次 await 是没有办法阻塞线程的,不能使用线程隔离的变量，要不怎么实现线程之间的通知，所以 ThreadLocal 是不可行的
     * private ThreadLocal<CountDownLatch> countDownLatch = ThreadLocal.withInitial(() -> new CountDownLatch(1));
     */
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    private static final String IP_PORT = "10.211.55.3:2181";

    /**
     * 根节点路径
     */
    private static final String ROOT_NODE = "/LOCK";

    /**
     * 当前节点的前置节点路径
     */
    private ThreadLocal<String> beforeNodePath = new ThreadLocal<>();

    /**
     * 当前节点路径
     */
    private ThreadLocal<String> nodePath = new ThreadLocal<>();

    private ZkClient zkClient = new ZkClient(IP_PORT);

    public ZkDistributedLock() {

        // 创建分布式锁对象时，初始化 zk 的路径
        if (!zkClient.exists(ROOT_NODE)) {
            zkClient.createPersistent(ROOT_NODE);
        }

    }

    /**
     * 加锁方法
     */
    @Override
    public void lock() {
        if (tryLock()) {
            log.info("{} 加锁成功", Thread.currentThread().getName());
            return;
        }
        // 阻塞 - 等待下次加锁的时机
        waitForLock();
        // 再次尝试加锁
        lock();
    }

    /**
     * 阻塞 - 等待下次加锁的时机
     */
    private void waitForLock() {

        // 监听前置节点的删除事件 - 监听内部类
        IZkDataListener zkDataListener = new IZkDataListener() {

            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {

            }

            /**
             * 监听节点的删除时间
             * @param dataPath 节点路径
             * @throws Exception 异常
             */
            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                log.info("{} 前置节点被删除", dataPath);
                // TODO: 2021/11/24   这个countDown 会将所有正在等待的线程都唤醒，没有实现只唤醒自己的后置节点
                countDownLatch.countDown();
            }

        };

        // 订阅监听前置节点的删除时间
        zkClient.subscribeDataChanges(beforeNodePath.get(), zkDataListener);
        // 判断在监听之前，前置节点是否已经被删除
        if (zkClient.exists(beforeNodePath.get())) {
            // 前置节点还存在 - 阻塞线程 等待前置节点被删除之后继续执行
            try {
                countDownLatch.await();
                log.info("阻塞线程: {}, nodePath: {}", Thread.currentThread().getName(), nodePath.get());
            } catch (InterruptedException e) {
                log.info("阻塞 {} 线程失败, 中断此线程", Thread.currentThread().getName());
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }
        }
        // 重置 countDownLatch , 前置节点已经被删除，取消订阅事件
        countDownLatch = new CountDownLatch(1);
        zkClient.unsubscribeDataChanges(beforeNodePath.get(), zkDataListener);

    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    /**
     * 尝试加锁
     * @return boolean: 是否加锁成功
     */
    @Override
    public boolean tryLock() {

        // 1. 判断 nodePath 是否为空，为空的话说明 ZkDistributedLock 第一次申请锁， zk 需要进行临时节点的创建
        if (!StringUtils.hasText(nodePath.get())) {
            nodePath.set(zkClient.createEphemeralSequential(ROOT_NODE + "/", "lock"));
            // log.info("ZkDistributedLock 第一次申请锁， zk 需要进行临时节点的创建：{}", nodePath);
            log.info("nodePath为空,创建临时节点：{}", nodePath.get());
        }

        // 2. 获取 根节点所有子节点
        List<String> childrenNodeList = zkClient.getChildren(ROOT_NODE);

        // 3. 将子节点列表进行排序
        Collections.sort(childrenNodeList);

        log.info("nodePath: {}, nodeList:{}", nodePath.get(), childrenNodeList);

        // 4. 判断当前线程是为最小的节点，是最小的节点说明获取锁成功，反之等待并监听自己前面的节点，当自己前面的节点删除之后，就是自己再次申请锁的时候
        if (nodePath.get().equals(ROOT_NODE + "/" + childrenNodeList.get(0))) {
            log.info("线程名称: {}, nodePath: {} 是最小的节点，获取锁成功。", Thread.currentThread().getName(), nodePath.get());
            return true;
        } else {
            // 获取当前节点应该在 节点列表中插入的位置 进而取得他的上一个节点
            int i = Collections.binarySearch(childrenNodeList, nodePath.get().substring(ROOT_NODE.length() + 1));
            // 获取上一个节点的路径
            beforeNodePath.set(ROOT_NODE + "/" + childrenNodeList.get(i - 1));
            log.info("{} 前面的节点为：{}", nodePath.get(), beforeNodePath.get());
        }

        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        zkClient.delete(nodePath.get());
    }

    @Override
    public Condition newCondition() {
        return null;
    }

}

代码解析：

1. implements Lock: 实现 Lock 遵循 JUC 提供的规范。
2. 使用 zkClient.createEphemeralSequential() 创建临时顺序节点，避免羊群效应和拥有锁线程意外挂掉造成死锁的问题。
3. 阻塞线程这边使用的是 CountDownLatch。当有线程争抢到锁之后，其他的线程会被 CountDownLatch 的 await 方法给阻塞，当被阻塞线程的前置节点被删除，就说明当前节点应该被唤醒，因为顺序节点是有序的，所以只唤醒当前节点就可以了。这边唤醒方法使用的是 zkClient.subscribeDataChanges(beforeNodePath.get(), zkDataListener); 在检测到前置节点被删除之后使用 CountDownLatch 的 countDown 方法，当 countDown() 变成 0 之后就会唤醒线程。
4. nodePath 和 beforeNodePath 应该是线程私有变量，这样才能保证，每个线程记录自己的 nodePath 和 beforeNodePath。
5. 具体的实现细节可以参考代码中的注释。

步骤 3 问题说明：

1. 当 countDown() 变成 0 之后就会唤醒所有的线程，这边应该实现成唤醒自己下一个节点
2. 当 countDwon() 变成 0 之后需要重新 new 这个对象，不然 await 方法是没有办法重新阻塞线程的。

测试自定义 zk 分布式锁：

java复制代码package com.aha.lock.zk;

import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author: WT
 * @date: 2021/11/23 17:56
 */
@RestController
@Slf4j
public class ZkDistributeLockTest {

    static int inventory = 10;
    private static final int NUM = 10;

    private final ZkDistributedLock zkDistributedLock = new ZkDistributedLock();

    @GetMapping("/zk/lock")
    public void zkLockTest() {
        try {
            for (int i = 0; i < NUM; i++) {
                new Thread(() -> {
                        try {
                            zkDistributedLock.lock();
                            Thread.sleep(200);
                            if (inventory > 0) {
                                inventory--;
                            }
                            log.warn("库存扣减完之后为：{}", inventory);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            Thread.currentThread().interrupt();
                        } finally {
                            zkDistributedLock.unlock();
                        }
                    }
                ).start();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

使用 curator 的 zk 分布式锁

curator 提供的几种分布式锁方案

1. InterProcessMutex：分布式可重入排它锁
2. InterProcessSemaphoreMutex：分布式排它锁
3. InterProcessReadWriteLock：分布式读写锁

InterProcessMutex 使用实例

配置 curatorFramework 客户端

yaml复制代码zookeeper:
  address: 10.211.55.3:2181   # zookeeper Server 地址,如果有多个使用逗号分隔。如 ip1:port1,ip2:port2,ip3:port3
  retryCount: 5               # 重试次数
  initElapsedTimeMs: 1000     # 初始重试间隔时间
  maxElapsedTimeMs: 5000      # 最大重试间隔时间
  sessionTimeoutMs: 30000     # Session 超时时间
  connectionTimeoutMs: 10000  # 连接超时时间

java复制代码package com.aha.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

/**
 * 连接 zookeeper 配置类
 *
 * @author: WT
 * @date: 2021/11/22 18:14
 */
@Data
@Configuration
@ConfigurationProperties(prefix = "zookeeper")
public class ZkClientProperties {

    /** 重试次数 */
    private int retryCount;

    /** 初始重试间隔时间 */
    private int initElapsedTimeMs;

    /** 最大重试间隔时间 */
    private int maxElapsedTimeMs;

    /**连接地址 */
    private String address;

    /**Session过期时间 */
    private int sessionTimeoutMs;

    /**连接超时时间 */
    private int connectionTimeoutMs;

}

java复制代码package com.aha.client;

import com.aha.config.ZkClientProperties;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 生成 zk 客户端
 *
 * @author: WT
 * @date: 2021/11/22 18:18
 */
@Configuration
public class ZookeeperClient {

    /**
     * initMethod = "start"
     * curatorFramework 创建对象之后，调用 curatorFramework 实例的 start 方法
     */
//    @Bean(initMethod = "start")
//    public CuratorFramework curatorFramework(ZkClientProperties zookeeperProperties) {
//        return CuratorFrameworkFactory.newClient(
//                zookeeperProperties.getAddress(),
//                zookeeperProperties.getSessionTimeoutMs(),
//                zookeeperProperties.getConnectionTimeoutMs(),
//                new RetryNTimes(zookeeperProperties.getRetryCount(), zookeeperProperties.getInitElapsedTimeMs())
//        );
//    }

    @Bean(initMethod = "start")
    private static CuratorFramework getZkClient(ZkClientProperties zookeeperProperties) {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, zookeeperProperties.getRetryCount(), 5000);
        return CuratorFrameworkFactory.builder()
                .connectString(zookeeperProperties.getAddress())
                .sessionTimeoutMs(zookeeperProperties.getSessionTimeoutMs())
                .connectionTimeoutMs(zookeeperProperties.getConnectionTimeoutMs())
                .retryPolicy(retryPolicy)
                .build();
    }

}

模拟 50 个线程争抢锁：

java复制代码package com.aha.lock.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.springframework.stereotype.Service;

/**
 *
 * @author: WT
 * @date: 2021/11/22 17:41
 */
@Slf4j
@Service
public class InterprocessMutexLock {

    private final CuratorFramework curatorFramework;

    public InterprocessMutexLock (CuratorFramework curatorFramework) {
        this.curatorFramework = curatorFramework;
    }

    public void test(String lockPath)  {

        InterProcessMutex lock = new InterProcessMutex(curatorFramework, lockPath);
        //模拟 50 个线程抢锁
        for (int i = 0; i < 50; i++) {
            new Thread(new TestThread(i, lock)).start();
        }

    }

    static class TestThread implements Runnable {

        private final Integer threadFlag;
        private final InterProcessMutex lock;

        public TestThread(Integer threadFlag, InterProcessMutex lock) {
            this.threadFlag = threadFlag;
            this.lock = lock;
        }

        @Override
        public void run() {
            try {
                lock.acquire();
                log.info("第 {} 个线程获取到了锁", threadFlag);
                //等到1秒后释放锁
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                try {
                    lock.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

}

java复制代码package com.aha.lock.controller;

import com.aha.lock.service.InterprocessMutexLock;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author: WT
 * @date: 2021/11/23 14:13
 */
@RestController
public class TestController {

    private final InterprocessMutexLock interprocessMutexLock;

    public TestController (InterprocessMutexLock interprocessMutexLock) {
        this.interprocessMutexLock = interprocessMutexLock;
    }

    @GetMapping("/lock/mutex")
    public void testMutexLock () {
        interprocessMutexLock.test("/lock/mutex");
    }


}

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有