原 荐 缓存架构SpringBoot集成Curator实现zookeeper分布式锁

缓存架构SpringBoot集成Curator实现zookeeper分布式锁

这篇文章其实是下篇文章 缓存架构之实战演练基于zk分布式锁解决分布式缓存并发冲突问题 做理论铺垫的，下篇文章我们就会利用该 工具 解决分布式缓存并发冲突问题，就是下面这个架构，下篇文章我们会重点讨论：

一、分布式锁简介

1、什么是锁

• 在单机环境下，当存在多个线程可以同时改变某个共享变量时，就需要同步来实现该功能，使其线程安全。
• 而同步就是通过锁来实现的。锁保证了同一时刻只有一个线程来修改共享变量。

在单机环境下，Java提供了一些并发安全包可以一定程度上保证线程安全，但是在分布式环境(多机环境)下，这些并发包显得就无能为力了！！

2、什么是分布式

分布式的CAP理论：

任何一个分布式系统都无法同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance），最多只能同时满足两项。

目前很多大型网站及应用都是分布式部署的，分布式场景中的数据一致性问题一直是一个比较重要的话题。基于 CAP理论，很多系统在设计之初就要对这三者做出取舍。在互联网领域的绝大多数的场景中，都需要牺牲强一致性来换取系统的高可用性，系统往往只需要保证最终一致性。

3、什么是分布式锁

顾名思义，分布式锁肯定是用在分布式环境下。在分布式环境下，使用分布式锁的目的也是保证同一时刻只有一个线程来修改共享变量，修改共享缓存……。

下篇文章，我们将分享一个实战案例，就是： 缓存架构之实战演练基于zk分布式锁解决分布式缓存并发冲突问题

二、原生zookeeper实现分布式锁

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CountDownLatch;

/**
 * zookeeper工具类：
 *
 * 更多免费资料，更多高清内容，更多 java 技术，欢迎访问网站
 * 极客慧：www.jikeh.cn
 * 如果你希望进一步深入交流，请加入我们的大家庭QQ群：375412858
 */
public class ZooKeeperSession {

    private static final Logger logger = LoggerFactory.getLogger(ZooKeeperSession.class);

    private static CountDownLatch connectedSemaphore = new CountDownLatch(1);

    private ZooKeeper zookeeper;

    public ZooKeeperSession() {

        // 连接zookeeper server，是异步创建会话的，那我们怎么知道zk session建立成功了呢？
        // 通过一个监听器+CountDownLatch，来确认真正建立了zk server的连接
        try {
            this.zookeeper = new ZooKeeper(
                    "localhost:2181",
                    50000,
                    new ZooKeeperWatcher());

            //打印即使状态：验证其是不是异步的？
            logger.info(String.valueOf(zookeeper.getState()));

            try {
                // CountDownLatch：简而言之 初始化——非0；非0——等待；0——往下执行
                connectedSemaphore.await();
            } catch(InterruptedException e) {
                e.printStackTrace();
            }

            logger.info("ZooKeeper session established......");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 初始化实例：
     */
    public static void init() {
        getInstance();
    }

    /**
     * 建立zk session的watcher：
     */
    private class ZooKeeperWatcher implements Watcher {

        public void process(WatchedEvent event) {
            if(KeeperState.SyncConnected == event.getState()) {
                connectedSemaphore.countDown();
            }
        }

    }

    /**
     * 静态内部类实现单例：
     */
    private static class Singleton {

        private static ZooKeeperSession instance;

        static {
            instance = new ZooKeeperSession();
        }

        public static ZooKeeperSession getInstance() {
            return instance;
        }

    }

    /**
     * 获取单例：
     *
     * @return
     */
    public static ZooKeeperSession getInstance() {
        return Singleton.getInstance();
    }

    /**
     * 重试获取分布式锁：
     *
     * @param adId
     */
    public void acquireDistributedLock(Long adId) {
        String path = "/ad-lock-" + adId;

        try {
            zookeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            logger.info("success to acquire lock for adId = " + adId);
        } catch (Exception e) {
            // 如果那个广告对应的锁node，已经存在了，就是已经被别人加锁了，那么就这里就会报错
            // NodeExistsException
            int count = 0;
            while(true) {
                try {
                    Thread.sleep(1000);
                    zookeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                } catch (Exception e2) {
                    count++;
                    logger.info("the " + count + " times try to acquire lock for adId = " + adId);
                    continue;
                }
                logger.info("success to acquire lock for adId = " + adId + " after " + count + " times try......");
                break;
            }
        }
    }

    /**
     * 释放掉分布式锁：
     *
     * @param adId
     */
    public void releaseDistributedLock(Long adId) {
        String path = "/ad-lock-" + adId;
        try {
            zookeeper.delete(path, -1);
            logger.info("release the lock for adId = " + adId);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        Long adId = 1L;
        ZooKeeperSession zkSession = ZooKeeperSession.getInstance();
        //1、获取锁：
        zkSession.acquireDistributedLock(adId);

        //2、执行一些修改共享资源的操作
        logger.info("I am updating common resource！");

        //3、释放锁
        zkSession.releaseDistributedLock(adId);
    }

}

三、SpringBoot集成Curator实现zookeeper分布式锁

1、Curator简介

Apache Curator是Netflix公司开源的一个Zookeeper客户端，目前已经是Apache的顶级项目，与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量，通过封装的一套高级API，里面提供了更多丰富的操作，例如session超时重连、主从选举、分布式计数器、分布式锁等等适用于各种复杂场景的zookeeper操作。

2、SpringBoot集成Curator实现zk分布式锁

1）引入pom依赖

<!-- zookeeper -->
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.10</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </exclusion>
        <exclusion>
            <artifactId>log4j</artifactId>
            <groupId>log4j</groupId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.12.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>2.12.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
    </exclusions>
</dependency>

2）基本配置

#重试次数
curator.retryCount=5
#重试间隔时间
curator.elapsedTimeMs=5000
# zookeeper 地址
curator.connectString=127.0.0.1:2181
# session超时时间
curator.sessionTimeoutMs=60000
# 连接超时时间
curator.connectionTimeoutMs=5000

3）连接配置

package com.jikeh.config;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class CuratorConfiguration {

    @Value("${curator.retryCount}")
    private int retryCount;

    @Value("${curator.elapsedTimeMs}")
    private int elapsedTimeMs;

    @Value("${curator.connectString}")
    private String connectString;

    @Value("${curator.sessionTimeoutMs}")
    private int sessionTimeoutMs;

    @Value("${curator.connectionTimeoutMs}")
    private int connectionTimeoutMs;

    @Bean(initMethod = "start")
    public CuratorFramework curatorFramework() {
        return CuratorFrameworkFactory.newClient(
                connectString,
                sessionTimeoutMs,
                connectionTimeoutMs,
                new RetryNTimes(retryCount, elapsedTimeMs));
    }
}

4）Curator实现zk分布式锁工具类

package com.jikeh.lock;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.CountDownLatch;

@Service
public class DistributedLockByCurator implements InitializingBean{

    private static final Logger logger = LoggerFactory.getLogger(DistributedLockByCurator.class);

    private final static String ROOT_PATH_LOCK = "rootlock";
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    @Autowired
    private CuratorFramework curatorFramework;

    /**
     * 获取分布式锁
     */
    public void acquireDistributedLock(String path) {
        String keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
        while (true) {
            try {
                curatorFramework
                        .create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(keyPath);
                logger.info("success to acquire lock for path:{}", keyPath);
                break;
            } catch (Exception e) {
                logger.info("failed to acquire lock for path:{}", keyPath);
                logger.info("while try again .......");
                try {
                    if (countDownLatch.getCount() <= 0) {
                        countDownLatch = new CountDownLatch(1);
                    }
                    countDownLatch.await();
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    /**
     * 释放分布式锁
     */
    public boolean releaseDistributedLock(String path) {
        try {
            String keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
            if (curatorFramework.checkExists().forPath(keyPath) != null) {
                curatorFramework.delete().forPath(keyPath);
            }
        } catch (Exception e) {
            logger.error("failed to release lock");
            return false;
        }
        return true;
    }

    /**
     * 创建 watcher 事件
     */
    private void addWatcher(String path) throws Exception {
        String keyPath;
        if (path.equals(ROOT_PATH_LOCK)) {
            keyPath = "/" + path;
        } else {
            keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
        }
        final PathChildrenCache cache = new PathChildrenCache(curatorFramework, keyPath, false);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener((client, event) -> {
            if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                String oldPath = event.getData().getPath();
                logger.info("success to release lock for path:{}", oldPath);
                if (oldPath.contains(path)) {
                    //释放计数器，让当前的请求获取锁
                    countDownLatch.countDown();
                }
            }
        });
    }

    //创建父节点，并创建永久节点
    @Override
    public void afterPropertiesSet() {
        curatorFramework = curatorFramework.usingNamespace("lock-namespace");
        String path = "/" + ROOT_PATH_LOCK;
        try {
            if (curatorFramework.checkExists().forPath(path) == null) {
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(path);
            }
            addWatcher(ROOT_PATH_LOCK);
            logger.info("root path 的 watcher 事件创建成功");
        } catch (Exception e) {
            logger.error("connect zookeeper fail，please check the log >> {}", e.getMessage(), e);
        }
    }
}

5）测试控制器

• 首先访问链接（线程1）：http://localhost:1111/curator/lock1 首先拿到锁，锁保持20s，操作，放锁

• 再访问链接（线程2）：http://localhost:1111/curator/lock2 等待获取锁，锁保持15s，操作，放锁

• 结果分析

注释：红色——线程1执行结果；蓝色——线程2执行结果；

代码下载地址： https://gitee.com/jikeh/JiKeHCN-RELEASE.git 项目名 ：spring-boot-curator

更多内容，请关注：

头条号：极客慧

个人网站： 极客慧

更多资料分享，请入群讨论：375412858