Curator Zookeeper分布式锁

pom.xml中添加如下配置

    
      
     
      org.apache.curator
       
     
      curator-recipes
       
     
      2.10.0
     
    

zookeeper配置

下载zookeeper并解压至D:\java\zookeeper-3.4.6：

http://www.eu.apache.org/dist/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz

zookeeper配置文件：

zoo-1.cfg

# The number of milliseconds of each ticktickTime=2000# The number of ticks that the initial# synchronization phase can takeinitLimit=10# The number of ticks that can pass between# sending a request and getting an acknowledgementsyncLimit=5# the directory where the snapshot is stored.# do not use /tmp for storage, /tmp here is just# example sakes.dataDir=D:/java/zookeeper-3.4.6/data/1#日志位置dataLogDir=D:/java/zookeeper-3.4.6/log/1# the port at which the clients will connectclientPort=2181# the maximum number of client connections.# increase this if you need to handle more clients#maxClientCnxns=60## Be sure to read the maintenance section of the# administrator guide before turning on autopurge.## http:/zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance## The number of snapshots to retain in dataDir#autopurge.snapRetainCount=3# Purge task interval in hours# Set to "0" to disable auto purge feature#autopurge.purgeInterval=1server.1=localhost:2887:3887server.2=localhost:2888:3888server.3=localhost:2889:3889

zoo-2.cfg和zoo-3.cfg修改如下配置并创建相应的目录

修改clientPort：

zoo-1.cfg：clientPort=2181zoo-2.cfg：clientPort=2182zoo-3.cfg：clientPort=2183

创建目录：

zoo-1.cfg：D:/java/zookeeper-3.4.6/data/1zoo-2.cfg：D:/java/zookeeper-3.4.6/data/2zoo-3.cfg：D:/java/zookeeper-3.4.6/data/3

分别创建文件：myid，内容分别为各自的id：1、2和3

D:/java/zookeeper-3.4.6/data/1/myid：1D:/java/zookeeper-3.4.6/data/2/myid：2D:/java/zookeeper-3.4.6/data/3/myid：3

分别自动各个zookeeper实例

代码测试

import org.apache.curator.RetryPolicy;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.recipes.locks.InterProcessMutex;import org.apache.curator.retry.ExponentialBackoffRetry;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class CuratorLockTest {    public static void main(String[] args) throws InterruptedException {        CountDownLatch latch = new CountDownLatch(5);        String zookeeperConnectionString = "localhost:2181,localhost:2182,localhost:2183";        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);        CuratorFramework client = CuratorFrameworkFactory.newClient(                zookeeperConnectionString, retryPolicy);        client.start();        System.out.println("客户端启动。。。。");        ExecutorService exec = Executors.newCachedThreadPool();        for (int i = 0; i < 5; i++) {            exec.submit(new MyLock("client" + i, client, latch));        }        exec.shutdown();        latch.await();        System.out.println("所有任务执行完毕");        client.close();        System.out.println("客户端关闭。。。。");    }    static class MyLock implements Runnable {        private String name;        private CuratorFramework client;        private CountDownLatch latch;        public MyLock(String name, CuratorFramework client, CountDownLatch latch) {            this.name = name;            this.client = client;            this.latch = latch;        }        public String getName() {            return name;        }        public void setName(String name) {            this.name = name;        }        public void run() {            InterProcessMutex lock = new InterProcessMutex(client, "/test_group");            try {                System.out.println("------" + this.name + "---------等待获取锁。--------");                if (lock.acquire(120, TimeUnit.SECONDS)) {                    try {                        System.out.println("----------" + this.name + "获得资源----------");                        System.out.println("----------" + this.name + "正在处理资源----------");                        Thread.sleep(10 * 1000);                        System.out.println("----------" + this.name + "资源使用完毕----------");                        latch.countDown();                    } finally {                        lock.release();                        System.out.println("----------" + this.name + "释放----------");                    }                }            } catch (Exception e) {                e.printStackTrace();            }        }    }}

运行结果：

客户端启动。。。。------client1---------等待获取锁。--------------client2---------等待获取锁。--------------client0---------等待获取锁。--------------client4---------等待获取锁。--------------client3---------等待获取锁。------------------client1获得资源--------------------client1正在处理资源--------------------client1资源使用完毕--------------------client1释放--------------------client3获得资源--------------------client3正在处理资源--------------------client3资源使用完毕--------------------client3释放--------------------client0获得资源--------------------client0正在处理资源--------------------client0资源使用完毕--------------------client0释放--------------------client4获得资源--------------------client4正在处理资源--------------------client4资源使用完毕--------------------client4释放--------------------client2获得资源--------------------client2正在处理资源--------------------client2资源使用完毕----------所有任务执行完毕

参考文档：