分布式锁--zookeeper--curator

curator具体应用,创建工具类,在项目启动时就加载该工具类,初始化CuratorFramework对象,在真正操作zk的地方,通过该工具类获取CuratorFramework对象

public class ZkClientUtil {
    private static final int BASE_SLEEP_TIME_MS = 5000; //定义失败重试间隔时间 单位:毫秒
    private static final int MAX_RETRIES = 3; //定义失败重试次数
    private static final int SESSION_TIME_OUT = 1000000; //定义会话存活时间,根据业务灵活指定 单位:毫秒
    private static final String ZK_URI = "192.168.134.131:2181,192.168.134.132:2181,192.168.134.133:2181";//你自己的zkurl
    // 和端口号
    private static final String NAMESPACE = "lock/aaa";
    //工作空间,可以不指定,建议指定,功能类似于项目包,之后创建的所有的节点都会在该工作空间下,方便管理
    private static CuratorFramework client;

    public static void init() {
        build();
        //注意要启动
        client.start();
    }

    public static CuratorFramework getClient() {
        return client;
    }

    public static InterProcessMutex getLock(String lockName){
        return new InterProcessMutex(client, lockName);
    }

    public static void build() {
        //创建比较简单,链式编程,很爽,基本上指定点参数就OK了
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME_MS, MAX_RETRIES);//重试策略
        client = CuratorFrameworkFactory
                .builder()
                .connectString(ZK_URI)
                .retryPolicy(retryPolicy)
                .namespace(NAMESPACE)
                .sessionTimeoutMs(SESSION_TIME_OUT)
                .build();
    }
}
public class ProductService {
    public final static int MAX_NUM = 20;
    public final static int MIN_NUM = 5;
    public volatile Integer productNum = 10;
    public final String lock = "/111";
    public final static ReentrantLock LOCK = new ReentrantLock();
    public final static Condition CONDITION = LOCK.newCondition();

    public void add() throws Exception {
        InterProcessMutex zkLock = ZkClientUtil.getLock(lock);
        if (!zkLock.acquire(500, TimeUnit.SECONDS)) {
            return;
        }
//        if(zkLock.acquire(50,TimeUnit.MILLISECONDS)){
//            System.out.println(Thread.currentThread().getName() + "第二次获取锁成功!");
//        }
        System.out.println(Thread.currentThread().getName() + "获取锁成功!");
        try {
            productNum++;
            System.out.println(Thread.currentThread().getName() + "生产1,当前总数量:" + productNum);
            TimeUnit.SECONDS.sleep(10);
        } catch (Exception e) {

        } finally {
            zkLock.release();
        }
        System.out.println("---------------------------------");
    }

    public void sub() throws Exception {
        InterProcessMutex zkLock = ZkClientUtil.getLock(lock);
        if (!zkLock.acquire(500, TimeUnit.SECONDS)) {
            return;
        }
//        if(zkLock.acquire(50,TimeUnit.SECONDS)){
//            System.out.println(Thread.currentThread().getName() + "第二次获取锁成功!");
//        }
        System.out.println(Thread.currentThread().getName() + "获取锁成功!");
        try {
            if (productNum > 0) {
                productNum--;
                System.out.println(Thread.currentThread().getName() + "消费1,剩余总数量:" + productNum);
            }

        } catch (Exception e) {

        } finally {
            zkLock.release();
        }

        System.out.println("---------------------------------");
    }
}

curator主要实现了下面四种锁

InterProcessMutex

​ 可重入排它锁,每成功加锁一次,就要解锁一次。

​ InterProcessMutex通过在zookeeper的某路径节点下创建临时序列节点来实现分布式锁,即每个线程获取同一把锁前,都需要在同样的路径下创建一个节点,节点名字由uuid + 递增序列组成。而通过对比自身的序列数是否在所有子节点的第一位,来判断是否成功获取到了锁。当获取锁失败时,它会添加watcher来监听前一个节点的变动情况,然后进入等待状态。直到watcher的事件生效将自己唤醒,或者超时时间异常返回。

InterProcessSemaphoreMutex

​ 不可重入排它锁
InterProcessReadWriteLock

​ 可重入读写锁,读共享,写互斥,一个拥有写锁的线程可重入读锁,但是读锁却不能进入写锁。这也意味着写锁可以降级成读锁, 比如请求写锁 —>请求读锁—>释放读锁 —->释放写锁
InterProcessMultiLock

​ 将多个锁作为单个实体管理的容器,当调用acquire(), 所有的锁都会被acquire(),如果请求失败,所有的锁都会被release。 同样调用release时所有的锁都被release(失败被忽略)。

©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。