分布式锁

基于Zookeeper的分布式锁

/
 * Zookeeper客户端
 *
 * @author lzlg
 * 2022/11/27 19:49
 */
@Component
public final class ZookeeperClient {

    private ZooKeeper zooKeeper;

    /
     * 程序启动初始化ZooKeeper客户端
     */
    @PostConstruct
    public void init() {
        CountDownLatch cdl = new CountDownLatch(1);
        try {
            zooKeeper = new ZooKeeper("192.168.174.160:2181", 30_000, event -> {
                Watcher.Event.KeeperState state = event.getState();
                if (state.equals(SyncConnected) && event.getType().equals(None)) {
                    cdl.countDown();
                    System.out.println("初始化ZooKeeper完成");
                }
            });
            cdl.await();
        } catch (IOException | InterruptedException e) {
            // 初始化失败,则抛出异常
            throw new IllegalStateException(e.getMessage());
        }
    }

    /
     * 程序停止销毁ZooKeeper客户端
     */
    @PreDestroy
    public void destroy() {
        if (!ObjectUtils.isEmpty(zooKeeper)) {
            try {
                zooKeeper.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /
     * 获取Zookeeper锁
     *
     * @param lockName 锁名称
     * @return Zookeeper锁
     */
    public ZookeeperLock getLock(String lockName) {
        return new ZookeeperLock(zooKeeper, lockName);
    }
}

/
 * Zookeeper分布式锁
 * @author lzlg
 * 2022/11/27 19:49
 */
public final class ZookeeperLock implements Lock {

    private final ZooKeeper zooKeeper;

    private final String lockName;

    private String currentNodePath;

    // 锁的根路径
    private static final String ROOT_PATH = "/locks";

    private static final ThreadLocal<Integer> renewCount = new ThreadLocal<>();

    ZookeeperLock(ZooKeeper zooKeeper, String lockName) {
        this.zooKeeper = Objects.requireNonNull(zooKeeper);
        this.lockName = Objects.requireNonNull(lockName);
        try {
            // 如果根路径节点不存在,则进行创建
            if (ObjectUtils.isEmpty(this.zooKeeper.exists(ROOT_PATH, false))) {
                // 创建根永久节点
                String result = this.zooKeeper.create(ROOT_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT);
                System.out.println("初始化根节点完成:" + result);
            }
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void lock() {
        this.tryLock();
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        throw new InterruptedException();
    }

    /
     * 加锁方法,如果已经存在LockName对应的节点,再次创建Zookeeper会创建失败
     * 注意,必须创建临时节点,如果使用永久节点,万一服务器宕机,则会发生死锁问题.
     *
     * 使用临时序列化节点+监听机制实现阻塞锁,不再递归调用
     */
    @Override
    public boolean tryLock() {
        // 如果已经有重入记录且重入次数大于0,则直接获取锁
        Integer flag = renewCount.get();
        if (!ObjectUtils.isEmpty(flag) && flag > 0) {
            renewCount.set(flag + 1);
            return true;
        }

        try {
            // 使用临时序列化节点,返回的是当前序列化节点路径
            currentNodePath = this.zooKeeper.create(this.lockPath(), null,
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println("获取锁成功,结果:" + currentNodePath);

            // 获取前一个节点名称
            Optional<String> preNodeOptional = this.getPreNode();
            // 如果有前一个节点
            if (preNodeOptional.isPresent()) {
                String preNode = preNodeOptional.get();
                CountDownLatch cdl = new CountDownLatch(1);
                // 判断前一个节点是否存在
                if (ObjectUtils.isEmpty(this.zooKeeper.exists(ROOT_PATH + "/" + preNode, event -> {
                    cdl.countDown();
                }))) {
                    // 如果节点不存在,则当前节点已经是第一个了,获取锁成功
                    renewCount.set(1);
                    return true;
                }
                // 如果有前一个节点,应该阻塞,直到前一个节点获取锁完成业务逻辑删除锁
                cdl.await();
            }
            // 返回空则是当前节点
            renewCount.set(1);
            return true;
        } catch (KeeperException | InterruptedException e) {
            System.out.println("加锁失败:" + e.getMessage());
            // 如果加锁失败,则进行重试
//            try {
//                TimeUnit.MILLISECONDS.sleep(60);
//                // 再次请求获取锁
//                this.tryLock();
//            } catch (InterruptedException ex) {
//                ex.printStackTrace();
//            }
        }
        return false;
    }

    /
     * 获取前一个节点名称
     *
     * @return 前节点名称
     */
    private Optional<String> getPreNode() {
        // 获取根路径下的子节点列表
        try {
            List<String> children = this.zooKeeper.getChildren(ROOT_PATH, false);
            // 如果没有子节点,则抛出异常,因为刚刚增加了一个临时序列节点
            if (ObjectUtils.isEmpty(children)) {
                throw new IllegalMonitorStateException("非法操作");
            }
            // 取出是相同前缀的子节点
            List<String> prefixSameNodes = children.stream().filter(n -> n.startsWith(lockName + "-"))
                    .collect(Collectors.toList());
            // 如果没有,则也是抛出异常
            if (ObjectUtils.isEmpty(prefixSameNodes)) {
                throw new IllegalMonitorStateException("非法操作");
            }
            // 进行排序
            Collections.sort(prefixSameNodes);
            // 找到当前节点的下标
            String currentNodeName = currentNodePath.substring(currentNodePath.lastIndexOf("/") + 1);
            int index = prefixSameNodes.indexOf(currentNodeName);
            // 如果当前节点下标小于0,则为非法操作
            if (index < 0) {
                throw new IllegalMonitorStateException("非法操作");
                // 如果当前节点大于0,则返回前一个节点名称
            } else if (index > 0) {
                // 获取前一个节点的名称
                return Optional.of(prefixSameNodes.get(index - 1));
            }
            return Optional.empty();
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
            throw new IllegalMonitorStateException("非法操作");
        }

    }

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

    /
     * 解锁方法,就是删除对应LockName的节点就可以了
     */
    @Override
    public void unlock() {
        Integer flag = renewCount.get();
        if (ObjectUtils.isEmpty(flag)) {
            throw new IllegalMonitorStateException("非法操作");
        }
        // 重入次数减1
        renewCount.set(flag - 1);
        // 如果重入次数为0,则删除锁
        if (renewCount.get() == 0) {
            try {
                this.zooKeeper.delete(currentNodePath, -1);
            } catch (InterruptedException | KeeperException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public Condition newCondition() {
        throw new IllegalArgumentException();
    }

    /
     * 获取锁路径
     *
     * @return 锁路径
     */
    private String lockPath() {
        return ROOT_PATH + "/" + lockName + "-";
    }
}

@Service
public class LockServiceImpl extends ServiceImpl<LockMapper, Lock> implements LockService {

    @Autowired
    private RedisStockService redisStockService;

    /
     * 扣减库存
     */
    @Override
    public void stockDeduct() {
        Lock lock = new Lock();
        // 字段lock_name是唯一索引
        lock.setLockName("mysql-lock");
        try {
            this.save(lock);
            // 插入成功,加锁成功,则进行业务处理
            redisStockService.stockDeduct();
            // 处理完成,进行解锁
            this.removeById(lock.getId());
        } catch (Exception e) {
            System.out.println("加锁失败:" + e.getMessage());
            // 如果插入出现异常,则进行重试
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
            this.stockDeduct();
        }
    }
}