Apache Curator Distributed Lock

Question

I know Apache Curator can do the distributed lock feature which is build on the top of zookeeper. It looks like very easy to use based on the document which is posted in the Apache Curator official website. For example:

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
CuratorFramework client = CuratorFrameworkFactory.newClient("host:ip",retryPolicy);
client.start();

InterProcessSemaphoreMutex lock = new InterProcessSemaphoreMutex(client, path);
if(lock.acquire(10, TimeUnit.SECONDS))
{
     try { /*do something*/ }
     finally { lock.release(); }
}

But what does the second parameter "path" of "InterProcessSemaphoreMutex" mean? It means "the path for the lock" based on API but what exactly is it? Can anybody give me an example?

If I have millions of locks, do I have to create millions of "path to the lock"? Is there any limit that the maximum number of locks(znodes) a zookeeper cluster has? Or can we remove this lock when a process releases it?

Answer 1

ZooKeeper presents what looks like a distributed file system. For any ZooKeeper operation, recipe, etc., you write "znodes" to a particular path and watch for changes. See here: http://zookeeper.apache.org/doc/trunk/zookeeperOver.html#Simple+API (regarding znodes).

For Curator recipes, it needs to know the base path you want to use to perform the recipe. For InterProcessSemaphoreMutex, the path is what every participant should use. i.e. Process 1 and Process 2 want to both contend for the lock. So, they both allocate InterProcessSemaphoreMutex instances with the same path, say "/my/lock". Think of the path as the lock identifier. In the same ZooKeeper cluster, you could have multiple locks by using different paths.

Hope this helps (disclaimer: I'm the main author of Curator).

Answer 2

Some examples about Reaper.

@Test
public void     testSomeNodes() throws Exception
{

  Timing                  timing = new Timing();
  ChildReaper             reaper = null;
  CuratorFramework        client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
  try
  {
    client.start();

    Random              r = new Random();
    int                 nonEmptyNodes = 0;
    for ( int i = 0; i < 10; ++i )
    {
      client.create().creatingParentsIfNeeded().forPath("/test/" + Integer.toString(i));
      if ( r.nextBoolean() )
      {
        client.create().forPath("/test/" + Integer.toString(i) + "/foo");
        ++nonEmptyNodes;
      }
    }

    reaper = new ChildReaper(client, "/test", Reaper.Mode.REAP_UNTIL_DELETE, 1);
    reaper.start();

    timing.forWaiting().sleepABit();

    Stat    stat = client.checkExists().forPath("/test");
    Assert.assertEquals(stat.getNumChildren(), nonEmptyNodes);
  }
  finally
  {
    CloseableUtils.closeQuietly(reaper);
    CloseableUtils.closeQuietly(client);
  }
}

Java Code Examples for org.apache.curator.framework.recipes.locks.Reaper