Curator典型应用场景之分布式锁

在分布式环境中 ，为了保证数据的一致性，经常在程序的某个运行点(例如，减库存操作或者流水号生成等)需要进行同步控制。以一个"流水号生成"的场景为例，普通的后台应用通常都是使用时间戳来生成流水号，但是在用户访问量很大的情况下，可能会出现并发问题。下面通过示例程序就演示一个典型的并发问题：

public static void main(String[] args) throws Exception {

    CountDownLatch down = new CountDownLatch(1);
    for (int i=0;i<10;i++){
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    down.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss|SSS");
                String orderNo = sdf.format(new Date());
                System.out.println("生成的订单号是："+orderNo);
            }
        }).start();
    }
    down.countDown();
}

程序运行，输出结果如下：

生成的订单号是：15:30:28|365
生成的订单号是：15:30:28|365
生成的订单号是：15:30:28|367
生成的订单号是：15:30:28|367
生成的订单号是：15:30:28|367
生成的订单号是：15:30:28|367
生成的订单号是：15:30:28|369
生成的订单号是：15:30:28|398
生成的订单号是：15:30:28|367
生成的订单号是：15:30:28|367

不难发现，生成的10个订单不少都是重复的，如果是实际的生产环境中,这显然没有满足我们的也无需求。究其原因，就是因为在没有进行同步的情况下,出现了并发问题。下面我们来看看如何使用Curator实现分布式锁功能。

Recipes实现的锁有五种
Shared Reentrant Lockf分布式可重入锁
官网地址：http://curator.apache.org/curator-recipes/shared-reentrant-lock.html
Shared Lock 分布式非可重入锁
官网地址：http://curator.apache.org/curator-recipes/shared-lock.html
Shared Reentrant Read Write Lock可重入读写锁
官网地址：http://curator.apache.org/curator-recipes/shared-reentrant-read-write-lock.html
Shared Semaphore共享信号量
官网地址：http://curator.apache.org/curator-recipes/shared-semaphore.html
Multi Shared Lock 多共享锁
官网地址：http://curator.apache.org/curator-recipes/multi-shared-lock.html

Shared Reentrant Lock(分布式可重入锁)
全局同步的可重入分布式锁，任何时刻不会有两个客户端同时持有该锁。Reentrant和JDK的ReentrantLock类似， 意味着同一个客户端在拥有锁的同时，可以多次获取，不会被阻塞。

相关的类
InterProcessMutex

使用
创建InterProcessMutex实例
InterProcessMutex提供了两个构造方法，传入一个CuratorFramework实例和一个要使用的节点路径，InterProcessMutex还允许传入一个自定义的驱动类，默认是使用StandardLockInternalsDriver。

public InterProcessMutex(CuratorFramework client, String path);
public InterProcessMutex(CuratorFramework client, String path, LockInternalsDriver driver);

获取锁

使用acquire方法获取锁,acquire方法有两种:

public void acquire() throws Exception;

获取锁，一直阻塞到获取到锁为止。获取锁的线程在获取锁后仍然可以调用acquire() 获取锁(可重入)。 锁获取使用完后，调用了几次acquire(),就得调用几次release()释放。

public boolean acquire(long time, TimeUnit unit) throws Exception;

与acquire()类似，等待time * unit时间获取锁，如果仍然没有获取锁，则直接返回false。

释放锁
使用release(）方法释放锁
线程通过acquire()获取锁时，可通过release()进行释放，如果该线程多次调用 了acquire()获取锁，则如果只调用 一次release()该锁仍然会被该线程持有。

注意：同一个线程中InterProcessMutex实例是可重用的，也就是不需要在每次获取锁的时候都new一个InterProcessMutex实例，用同一个实例就好。

锁撤销
InterProcessMutex 支持锁撤销机制，可通过调用makeRevocable()将锁设为可撤销的，当另一线程希望你释放该锁时，实例里的listener会被调用。 撤销机制是协作的。

public void makeRevocable(RevocationListener<T> listener);

如果你请求撤销当前的锁， 调用Revoker类中的静态方法attemptRevoke()要求锁被释放或者撤销。如果该锁上注册有RevocationListener监听，该监听会被调用。

public static void attemptRevoke(CuratorFramework client, String path) throws Exception;

示例代码(官网)
共享资源

public class FakeLimitedResource {

    //总共250张火车票
    private Integer ticket = 250;

    public void use() throws InterruptedException {
        try {
            System.out.println("火车票还剩"+(--ticket)+"张！");
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

使用锁操作资源

public class ExampleClientThatLocks {

    /** 锁 */
    private final InterProcessMutex lock;
    /** 共享资源 */
    private final FakeLimitedResource resource;
    /** 客户端名称 */
    private final String clientName;

    public ExampleClientThatLocks(CuratorFramework client, String lockPath, FakeLimitedResource resource, String clientName) {
        this.resource = resource;
        this.clientName = clientName;
        lock = new InterProcessMutex(client, lockPath);
    }

    public void doWork(long time, TimeUnit unit) throws Exception {
        if ( !lock.acquire(time, unit) ) {
            throw new IllegalStateException(clientName + " could not acquire the lock");
        }
        try {
            System.out.println(clientName + " has the lock");
            //操作资源
            resource.use();
        } finally {
            System.out.println(clientName + " releasing the lock");
            lock.release(); //总是在Final块中释放锁。
        }
    }
}

客户端

public class LockingExample {
    private static final int QTY = 5;
    private static final int REPETITIONS = QTY * 10;
    private static final String CONNECTION_STRING = "172.20.10.9:2181";
    private static final String PATH = "/examples/locks";

    public static void main(String[] args) throws Exception {

        //FakeLimitedResource模拟某些外部资源，这些外部资源一次只能由一个进程访问
        final FakeLimitedResource resource = new FakeLimitedResource();

        ExecutorService service = Executors.newFixedThreadPool(QTY);
        try {
            for ( int i = 0; i < QTY; ++i ){
                final int index = i;
                Callable<Void>  task = new Callable<Void>() {
                    @Override
                    public Void call() throws Exception {
                        CuratorFramework client = CuratorFrameworkFactory.newClient(CONNECTION_STRING, new ExponentialBackoffRetry(1000, 3,Integer.MAX_VALUE));
                        try {
                            client.start();
                            ExampleClientThatLocks example = new ExampleClientThatLocks(client, PATH, resource, "Client " + index);
                            for ( int j = 0; j < REPETITIONS; ++j ) {
                                example.doWork(10, TimeUnit.SECONDS);
                            }
                        }catch ( InterruptedException e ){
                            Thread.currentThread().interrupt();
                        }catch ( Exception e ){
                            e.printStackTrace();
                        }finally{
                            CloseableUtils.closeQuietly(client);
                        }
                        return null;
                    }
                };
                service.submit(task);
            }

            service.shutdown();
            service.awaitTermination(10, TimeUnit.MINUTES);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

起五个线程，即五个窗口卖票，五个客户端分别有50张票可以卖，先是尝试获取锁，操作资源后，释放锁。

Shared Lock(不可重入锁)
与Shared Reentrant Lock类似,但是不能重入。

相关的类
InterProcessSemaphoreMutex

使用
创建InterProcessSemaphoreMutex实例

public InterProcessSemaphoreMutex(CuratorFramework client, String path);

示例代码
我们只需要将上面的例子修改一下，测试一下它的重入。 修改ExampleClientThatLocks,修改锁的类型，并连续两次acquire:

public class ExampleClientThatLocks {

    /** 锁 */
    private final InterProcessSemaphoreMutex lock;
    /** 共享资源 */
    private final FakeLimitedResource resource;
    /** 客户端名称 */
    private final String clientName;

    public ExampleClientThatLocks(CuratorFramework client, String lockPath, FakeLimitedResource resource, String clientName) {
        this.resource = resource;
        this.clientName = clientName;
        lock = new InterProcessSemaphoreMutex(client, lockPath);
    }

    public void doWork(long time, TimeUnit unit) throws Exception {
        if ( !lock.acquire(time, unit) ) {
            throw new IllegalStateException(clientName + " could not acquire the lock");
        }
        System.out.println(clientName + " has the lock");
        if ( !lock.acquire(time, unit) ) {
            throw new IllegalStateException(clientName + " could not acquire the lock");
        }
        System.out.println(clientName + " has the lock again");
        try {
            //操作资源
            resource.use();
        } finally {
            System.out.println(clientName + " releasing the lock");
            lock.release(); //总是在Final块中释放锁。
            lock.release(); //调用两次acquire释放两次
        }
    }
}

注意我们也需要调用release两次。这和JDK的ReentrantLock用法一致。如果少调用一次release，则此线程依然拥有锁。 上面的代码没有问题，我们可以多次调用acquire，后续的acquire也不会阻塞。

将上面的InterProcessMutex换成不可重入锁InterProcessSemaphoreMutex,如果再运行上面的代码，结果就会发现线程被阻塞再第二个acquire上。直到超时报异常：
java.lang.IllegalStateException: Client 1 could not acquire the lock 说明锁是不可重入的。

Shared Reentrant Read Write Lock分布式可重入读写锁
读写锁负责管理一对相关的锁，一个负责读操作，一个负责写操作。读锁在没有写锁没被使用时能够被多个读进行使用。但是写锁只能被一个进得持有。 只有当写锁释放时，读锁才能被持有，一个拥有写锁的线程可重入读锁，但是读锁却不能进入写锁。 这也意味着写锁可以降级成读锁， 比如请求写锁 —>读锁 —->释放写锁。 从读锁升级成写锁是不行的。可重入读写锁是“公平的”，每个用户将按请求的顺序获取锁。

相关的类
InterProcessReadWriteLock
InterProcessLock

使用
创建InterProcessReadWriteLock

public InterProcessReadWriteLock(CuratorFramework client, String basePath);

获取锁

可通过readLock()和writeLock())分别获取锁类型，再通过acquire()获取锁。

public InterProcessMutex readLock();
public InterProcessMutex writeLock();

示例代码

public class CuratorLockSharedReentrantReadWriteLockZookeeper {

    private static final int SECOND = 1000;
    private static final String PATH="/examples/locks";
    private static final String CONNECTION_STRING = "192.168.58.42:2181";

    public static void main(String[] args) throws Exception {

        CuratorFramework client = CuratorFrameworkFactory.newClient(CONNECTION_STRING, new ExponentialBackoffRetry(1000, 3,Integer.MAX_VALUE));
        client.start();
        // todo 在此可添加ConnectionStateListener监听
        System.out.println("Server connected...");
        final InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, PATH);
        final CountDownLatch down = new CountDownLatch(1);
        for (int i = 0; i < 30; i++) {
            final int index = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        down.await();
                        if (index % 2 == 0) {
                            lock.readLock().acquire();
                            SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss|SSS");
                            String orderNo = sdf.format(new Date());
                            System.out.println("[READ]生成的订单号是:" + orderNo);
                        } else {
                            lock.writeLock().acquire();
                            SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss|SSS");
                            String orderNo = sdf.format(new Date());
                            System.out.println("[WRITE]生成的订单号是:" + orderNo);
                        }

                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        try {
                            if (index % 2 == 0) {
                                lock.readLock().release();
                            } else {
                                lock.writeLock().release();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
        }
        // 保证所有线程内部逻辑执行时间一致
        down.countDown();
        Thread.sleep(10 * SECOND);
        if (client != null) {
            client.close();
        }
        System.out.println("Server closed...");
    }
}

运行程序,打印如下结果：

Server connected…
[WRITE]生成的订单号是:11:40:25|042
[WRITE]生成的订单号是:11:40:25|098
[READ]生成的订单号是:11:40:25|116
[READ]生成的订单号是:11:40:25|127
[READ]生成的订单号是:11:40:25|137
[READ]生成的订单号是:11:40:25|141
[READ]生成的订单号是:11:40:25|175
[READ]生成的订单号是:11:40:25|214
[WRITE]生成的订单号是:11:40:25|244
[READ]生成的订单号是:11:40:25|276
[READ]生成的订单号是:11:40:25|276
[WRITE]生成的订单号是:11:40:25|347
[WRITE]生成的订单号是:11:40:25|370
[READ]生成的订单号是:11:40:25|378
[WRITE]生成的订单号是:11:40:25|413
[WRITE]生成的订单号是:11:40:25|469
[WRITE]生成的订单号是:11:40:25|499
[WRITE]生成的订单号是:11:40:25|519
[READ]生成的订单号是:11:40:25|574
[WRITE]生成的订单号是:11:40:25|595
[WRITE]生成的订单号是:11:40:25|636
[WRITE]生成的订单号是:11:40:25|670
[READ]生成的订单号是:11:40:25|698
[WRITE]生成的订单号是:11:40:25|719
[WRITE]生成的订单号是:11:40:25|742
[READ]生成的订单号是:11:40:25|756
[READ]生成的订单号是:11:40:25|771
[READ]生成的订单号是:11:40:25|776
[WRITE]生成的订单号是:11:40:25|789
[READ]生成的订单号是:11:40:25|805
Server closed…

可以看到通过获得read锁生成的订单中是有重复的，而获取的写锁中是没有重复数据的。符合读写锁的特点。

共享信号量Shared Semaphore
一个计数的信号量类似JDK的Semaphore，所有使用相同锁定路径的jvm中所有进程都将实现进程间有限的租约。此外，这个信号量大多是“公平的” - 每个用户将按照要求的顺序获得租约。
有两种方式决定信号号的最大租约数。一种是由用户指定的路径来决定最大租约数，一种是通过SharedCountReader来决定。
如果未使用SharedCountReader，则不会进行内部检查比如A表现为有10个租约，进程B表现为有20个。因此，请确保所有进程中的所有实例都使用相同的numberOfLeases值。
acuquire()方法返回的是Lease对象，客户端在使用完后必须要关闭该lease对象(一般在finally中进行关闭)，否则该对象会丢失。如果进程session丢失（如崩溃），该客户端拥有的所有lease会被自动关闭，此时其他端能够使用这些lease。

相关的类
InterProcessSemaphoreV2
Lease
SharedCountReader

使用
创建实例

public InterProcessSemaphoreV2(CuratorFramework client, String path, int maxLeases);
public InterProcessSemaphoreV2(CuratorFramework client, String path, SharedCountReader count);

获取Lease
请求获取lease，如果Semaphore当前的租约不够，该方法会一直阻塞，直到最大租约数增大或者其他客户端释放了一个lease。 当lease对象获取成功后，处理完成后，客户端必须调用close该lease(可通过return()方法释放lease)。最好在finally块中close。

//获取一个租约
public Lease acquire() throws Exception;
//获取多个租约
public Collection<Lease> acquire(int qty) throws Exception;
//对应的有阻塞时间的acquire()方法
public Lease acquire(long time, TimeUnit unit) throws Exception;
public Collection<Lease> acquire(int qty, long time, TimeUnit unit) throws Exception;

释放lease

public void returnAll(Collection<Lease> leases);
public void returnLease(Lease lease);

示例代码

public class InterProcessSemaphoreExample {
    private static final int MAX_LEASE=10;
    private static final String PATH="/examples/locks";
    private static final String CONNECTION_STRING = "172.20.10.9:2181";

    public static void main(String[] args) throws Exception {
        FakeLimitedResource resource = new FakeLimitedResource();
        try{

            CuratorFramework client = CuratorFrameworkFactory.newClient(CONNECTION_STRING, new ExponentialBackoffRetry(1000, 3,Integer.MAX_VALUE));
            client.start();

            InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, PATH, MAX_LEASE);
            Collection<Lease> leases = semaphore.acquire(5);
            System.out.println("get " + leases.size() + " leases");
            Lease lease = semaphore.acquire();
            System.out.println("get another lease");

            resource.use();

            Collection<Lease> leases2 = semaphore.acquire(5, 10, TimeUnit.SECONDS);
            System.out.println("Should timeout and acquire return " + leases2);

            System.out.println("return one lease");
            semaphore.returnLease(lease);
            System.out.println("return another 5 leases");
            semaphore.returnAll(leases);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

构造参数中最多有10个租约，首先我们先获得了5个租约，然后再获取一个,这个时候semaphore还剩4个， 接着再请求了5个租约，因为semaphore还有4个租约，因为租约不够，阻塞到超时，还是没能满足，返回结果为null。

Multi Shared Lock(多共享分布式锁)
多个锁作为一个锁，可以同时在多个资源上加锁。一个维护多个锁对象的容器。当调用 acquire()时，获取容器中所有的锁对象，请求失败时，释放所有锁对象。同样调用release()也会释放所有的锁。

相关的类
InterProcessMultiLock
InterProcessLock

使用
创建InterProcessMultiLock

public InterProcessMultiLock(CuratorFramework client, List<String> paths);
public InterProcessMultiLock(List<InterProcessLock> locks);

使用方式和Shared Lock相同。

示例代码

public class InterProcessMultiLockExample {
    private static final String PATH1 = "/examples/locks1";
    private static final String PATH2 = "/examples/locks2";
    private static final String CONNECTION_STRING = "172.20.10.9:2181";

    public static void main(String[] args) throws Exception {
        FakeLimitedResource resource = new FakeLimitedResource();
        try {
            CuratorFramework client = CuratorFrameworkFactory.newClient(CONNECTION_STRING, new ExponentialBackoffRetry(1000, 3,Integer.MAX_VALUE));
            client.start();

            InterProcessLock lock1 = new InterProcessMutex(client, PATH1);
            InterProcessLock lock2 = new InterProcessSemaphoreMutex(client, PATH2);

            InterProcessMultiLock lock = new InterProcessMultiLock(Arrays.asList(lock1, lock2));

            if (!lock.acquire(10, TimeUnit.SECONDS)) {
                throw new IllegalStateException("could not acquire the lock");
            }
            System.out.println("has the lock");

            System.out.println("has the lock1: " + lock1.isAcquiredInThisProcess());
            System.out.println("has the lock2: " + lock2.isAcquiredInThisProcess());

            try {
                resource.use(); //操作资源
            } finally {
                System.out.println("releasing the lock");
                lock.release(); //在finally中释放锁
            }
            System.out.println("has the lock1: " + lock1.isAcquiredInThisProcess());
            System.out.println("has the lock2: " + lock2.isAcquiredInThisProcess());
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

新建一个InterProcessMultiLock， 包含一个重入锁和一个非重入锁。 调用acquire后可以看到线程同时拥有了这两个锁。 调用release看到这两个锁都被释放了。

统一的错误处理
看过官网的朋友一定发现,每一个锁的文章最下面都有一个Error Handling，内容直接一键翻译过来：
强烈建议您添加ConnectionStateListener并监视SUSPENDED和LOST状态更改。如果报告了SUSPENDED状态，则除非您随后收到RECONNECTED状态，否则您无法确定是否仍然持有该锁。如果报告了LOST状态，则确定您不再持有锁。

当连接出现异常, 将通过ConnectionStateListener接口进行监听, 并进行相应的处理, 这些状态变化包括:
暂停(SUSPENDED): 当连接丢失, 将暂停所有操作, 直到连接重新建立, 如果在规定时间内无法建立连接, 将触发LOST通知
重连(RECONNECTED): 连接丢失, 执行重连时, 将触发该通知
丢失(LOST): 连接超时时, 将触发该通知

新建一个类实现ConnectionStateListener

public class MyConnectionStateListener implements ConnectionStateListener {

    /** 节点路径 */
    private String zkRegPathPrefix;
    /** 节点内容 */
    private String regContent;

    public MyConnectionStateListener(String zkRegPathPrefix, String regContent) {
        this.zkRegPathPrefix = zkRegPathPrefix;
        this.regContent = regContent;
    }

    @Override
    public void stateChanged(CuratorFramework client, ConnectionState newState) {
        if (newState == ConnectionState.LOST) {
            //连接丢失
            System.out.println("lost session with zookeeper");
            System.out.println("锁已经释放,不再拥有该锁");
            while(true){
                try {
                    System.err.println("尝试重新连接......");
                    if(client.getZookeeperClient().blockUntilConnectedOrTimedOut()){
                        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(zkRegPathPrefix, regContent.getBytes("UTF-8"));
                        break;
                    }
                } catch (InterruptedException e) {
                    break;
                } catch (Exception e){
                    //TODO: log something
                }
            }
        } else if (newState == ConnectionState.CONNECTED) {
            //连接新建
            System.out.println("connected with zookeeper");
        } else if (newState == ConnectionState.RECONNECTED) {
            //重新连接
            System.out.println("reconnected with zookeeper");
        }
    }
}

在client中添加该监听

CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.58.42:2181",3000,3000, new ExponentialBackoffRetry(1000, 3,Integer.MAX_VALUE));
client.start();
// todo 在此可添加ConnectionStateListener监听
MyConnectionStateListener connectionStateListener = new MyConnectionStateListener(PATH,"123456");
client.getConnectionStateListenable().addListener(connectionStateListener);
System.out.println("Server connected...");

启动程序,然后断掉网络,就会触发监听，接收到ConnectionState.LOST状态,表明该客户端已经不再持有该锁。