Java 类org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex 实例源码

/**
 * Run the callable only if the lock can be obtained.
 * 
 * @param <T>
 * @param lockId
 * @param timout
 *            lock time out
 * @param unit
 *            TimeUnit
 * @param c
 * @return T return the object returned by the Callable
 * @throws Exception
 */
public <T> T withLock(String lockId, long timeout, TimeUnit unit,
        Callable<T> c) throws Exception {

    final String lockPath = calcLockPath(lockId);
    final InterProcessSemaphoreMutex mutex = new InterProcessSemaphoreMutex(
            connection.get(), lockPath);

    if (mutex.acquire(timeout, unit)) {
        try {
            return c.call();
        } finally {
            mutex.release();
        }
    }else {
        throw new RuntimeException("Could not attain lock for " + lockId);
    }

}

@Override
public void getLockWithTimeout(String name, long timeout, Handler<AsyncResult<Lock>> resultHandler) {
  vertx.executeBlocking(future -> {
    ZKLock lock = locks.get(name);
    if (lock == null) {
      InterProcessSemaphoreMutex mutexLock = new InterProcessSemaphoreMutex(curator, ZK_PATH_LOCKS + name);
      lock = new ZKLock(mutexLock);
    }
    try {
      if (lock.getLock().acquire(timeout, TimeUnit.MILLISECONDS)) {
        locks.putIfAbsent(name, lock);
        future.complete(lock);
      } else {
        future.fail(new VertxException("Timed out waiting to get lock " + name));
      }
    } catch (Exception e) {
      future.fail(new VertxException("get lock exception", e));
    }
  }, resultHandler);
}

@Override
public Boolean releaseLock(String lockHolder, String lockName, LockHandle lockHandle) {
    InterProcessSemaphoreMutex lock = locks.get(lockName);
    if (lock == null) {
        logger.debug(lockName + " lock was never obtained");
        // TODO maybe ignore as we dont really care
        throw new IllegalStateException("Cannot release " + lockName + " since lock does not exist (was never obtained?)");
    }

    try {
        logger.debug("releasing lock " + lockName);
        lock.release();
        return true;
    } catch (Exception e) {
        logger.debug("exception swallowed while trying to release lock " + lockName, e);
        logger.trace(ExceptionToString.format(e));
    }
    return false;
}

@Override
public void flush() {
    if (client != null) {
        InterProcessSemaphoreMutex l = CuratorUtils.acquireLock(client,
                "/partition-lock-" + this.filePath.hashCode());


        HDFSUtils.appendBytes(this.fs, this.filePath, this.buffer, 0,
                curPointer);
        this.curPointer = 0;

        CuratorUtils.releaseLock(l);

    } else {
        HDFSUtils.appendBytes(this.fs, this.filePath, this.buffer, 0,
                curPointer);
        this.curPointer = 0;
    }

}

private Cluster$(String connectString) {
    RetryPolicy policy = new ExponentialBackoffRetry(10000, 5) ;
    try {
        client = CuratorFrameworkFactory.builder()
            /*.namespace(namespace)*/
            .connectString(connectString)
            .retryPolicy(policy)
            .build() ;
        client.start() ;

        client.blockUntilConnected() ;

    }
    catch (Exception e) {
        log.error("Failed: "+connectString, e) ;
        client = null ;
    }
    ensure(ClusterCtl.namespace) ;
    ensure(ClusterCtl.members) ;
    active.set(true) ;
    globalCounter = new DistributedAtomicLong(client,"/COUNTER", policy) ;
    try {
        log.info("/COUNTER = "+globalCounter.get().postValue());
    } catch (Exception ex) {}
    globalWriteLock = new InterProcessSemaphoreMutex(client, "/WriteLock") ; 
}

@Test
public void testLockedResource() throws Exception {
    final String path = ZNODE_COORDINATION + "/" + md5Hex(locked.getIRIString());
    final InterProcessLock lock = new InterProcessSemaphoreMutex(getZkClient(curator.getConnectString()), path);
    assertTrue(lock.acquire(100L, MILLISECONDS));

    final Dataset dataset = rdf.createDataset();
    dataset.add(rdf.createQuad(Trellis.PreferUserManaged, locked, DC.title, rdf.createLiteral("A title")));

    final ResourceService svc = new MyResourceService(curator.getConnectString(), mockEventService, null);
    assertFalse(svc.put(locked, null, dataset).get());
    assertTrue(svc.put(resource, null, dataset).get());
    assertTrue(svc.put(existing, null, dataset).get());
    verify(mockEventService, times(2)).emit(any(Notification.class));
}

@Override
public void start() {
  try {
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);  
    client = CuratorFrameworkFactory.builder()  
            .connectString(getIdentifier().getConnectionString())
            .connectionTimeoutMs(10000)
            .retryPolicy(retryPolicy)
            //.namespace("text")
            .build();
    client.getConnectionStateListenable().addListener(new ZkConnectionStateListener());
    client.start();
    if (client.getZookeeperClient().blockUntilConnectedOrTimedOut()) {

      lock = new InterProcessSemaphoreMutex(client, buildBasePath());

      serializer = new JsonInstanceSerializer<HostMetadata>(HostMetadata.class);
      serviceDiscovery = ServiceDiscoveryBuilder.builder(HostMetadata.class)
          .basePath(buildBasePath()).client(client).serializer(serializer).build();
      //force to create a root path if the node is not exist.
      ensureNodeExists(buildBasePath());
    }
  } catch (Exception e) {
    log.error("Service registry start error for server identifier '{}' !",
        getIdentifier().getConnectionString(), e);
    throw new ServiceRegistryException(e);
  }
}

/**
 * Gets an exclusive lock on service-specific ZK node to ensure two schedulers aren't running simultaneously for the
 * same service.
 */
public void lock() {
    if (curatorClient != null) {
        throw new IllegalStateException("Already locked");
    }
    curatorClient = CuratorFrameworkFactory.newClient(zookeeperConnection, CuratorUtils.getDefaultRetry());
    curatorClient.start();

    final String lockPath = PersisterUtils.join(CuratorUtils.getServiceRootPath(serviceName), LOCK_PATH_NAME);
    InterProcessSemaphoreMutex curatorMutex = new InterProcessSemaphoreMutex(curatorClient, lockPath);

    LOGGER.info("Acquiring ZK lock on {}...", lockPath);
    final String failureLogMsg = String.format("Failed to acquire ZK lock on %s. " +
            "Duplicate service named '%s', or recently restarted instance of '%s'?",
            lockPath, serviceName, serviceName);
    try {
        // Start at 1 for pretty display of "1/3" through "3/3":
        for (int attempt = 1; attempt < LOCK_ATTEMPTS + 1; ++attempt) {
            if (curatorMutex.acquire(10, getWaitTimeUnit())) {
                LOGGER.info("{}/{} Lock acquired.", attempt, LOCK_ATTEMPTS);
                this.curatorMutex = curatorMutex;
                return;
            }
            if (attempt < LOCK_ATTEMPTS) {
                LOGGER.error("{}/{} {} Retrying lock...", attempt, LOCK_ATTEMPTS, failureLogMsg);
            }
        }
        LOGGER.error(failureLogMsg + " Restarting scheduler process to try again.");
    } catch (Exception ex) {
        LOGGER.error(String.format("Error acquiring ZK lock on path: %s", lockPath), ex);
    }
    curatorClient = null;
    exit();
}

private InterProcessSemaphoreMutex acquireMutexOrThrow(long timeout, TimeUnit timeoutTimeUnit, String lockPath) throws Exception {
    InterProcessSemaphoreMutex mutex = new InterProcessSemaphoreMutex(curator.framework(), lockPath);

    log.log(LogLevel.DEBUG, "Waiting for lock on " + lockPath);
    boolean acquired = mutex.acquire(timeout, timeoutTimeUnit);
    if (!acquired) {
        log.log(LogLevel.DEBUG, "Timed out waiting for lock on " + lockPath);
        throw new TimeoutException("Timed out waiting for lock on " + lockPath);
    }
    log.log(LogLevel.DEBUG, "Successfully acquired lock on " + lockPath);
    return mutex;
}

public HDFSBufferedOutputStream(FileSystem fs, String filePath,
                                short replication, int bufferSize, CuratorFramework client) {
    this.buffer = new byte[bufferSize];
    this.filePath = filePath;
    this.fs = fs;
    InterProcessSemaphoreMutex l = CuratorUtils.acquireLock(client,
            "/partition-lock-" + this.filePath.hashCode());
    // Create the file if it does not exist.
    HDFSUtils.safeCreateFile(this.fs, filePath, replication);
    CuratorUtils.releaseLock(l);
    this.client = client;
}

public static InterProcessSemaphoreMutex acquireLock(
        CuratorFramework client, String lockPath) {

    // make sure that the zookeeper client is connected
    int connTries = 0, maxConnTries = 100;
    while (!(client.getZookeeperClient().isConnected())
            && connTries < maxConnTries) {
        ThreadUtils.sleep(500);
        connTries++;
    }
    if (!(client.getZookeeperClient().isConnected()))
        throw new RuntimeException("The zookeeper cient is not connected!");

    InterProcessSemaphoreMutex lock = new InterProcessSemaphoreMutex(
            client, lockPath);

    try {
        if (lock.acquire(waitTimeSeconds, TimeUnit.SECONDS))
            return lock;
        else {
            System.out.println("Time out: Failed to obtain lock: "
                    + lockPath);
            // return null;
            throw new RuntimeException("Failed to obtain lock: " + lockPath);
        }
    } catch (Exception e) {
        e.printStackTrace();
        // return null;
        throw new RuntimeException("Failed to obtain lock: " + lockPath
                + "\n " + e.getMessage());
    }
}

public static void releaseLock(InterProcessSemaphoreMutex lock) {
    try {
        lock.release();
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException("Failed to unlock " + lock + "\n "
                + e.getMessage());
    }
}

@Override
public boolean load() {
    if (path == null || path.equals("")) {
        throw new RuntimeException();
    }
    try {
        totalSize = Math.min(800 * 1024 * 1024, totalSize); // 800 MB

        InterProcessSemaphoreMutex l = CuratorUtils.acquireLock(client,
                "/partition-lock-" + path.hashCode() + "-" + partitionId);

        System.out.println("LOCK: acquired lock,  " + "path=" + path
                + " , partition id=" + partitionId + " , for loading, size: " + totalSize);
        Path p = new Path(path + "/" + partitionId);
        in = hdfs.open(p);
        bytes = new byte[(int) totalSize];
        ByteStreams.readFully(in, bytes);
        in.close();

        CuratorUtils.releaseLock(l);
        return true; // load the physical block for this partition
    } catch (IOException e) {
        e.printStackTrace();
        throw new RuntimeException("Failed to read file: " + path + "/"
                + partitionId);
    }
}

public ZookeeperBasedJobLock(Properties properties) throws JobLockException {
  String jobName = properties.getProperty(ConfigurationKeys.JOB_NAME_KEY);
  this.lockAcquireTimeoutMilliseconds =
      getLong(properties, LOCKS_ACQUIRE_TIMEOUT_MILLISECONDS, LOCKS_ACQUIRE_TIMEOUT_MILLISECONDS_DEFAULT);
  this.lockPath = Paths.get(LOCKS_ROOT_PATH, jobName).toString();
  initializeCuratorFramework(properties);
  lock = new InterProcessSemaphoreMutex(curatorFramework, lockPath);
}

public static void main(String[] args) throws Exception {
    FakeLimitedResource resource = new FakeLimitedResource();
    try (TestingServer server = new TestingServer()) {
        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
        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(); //access resource exclusively
        } finally {
            System.out.println("releasing the lock");
            lock.release(); // always release the lock in a finally block
        }
        System.out.println("has the lock1: " + lock1.isAcquiredInThisProcess());
        System.out.println("has the lock2: " + lock2.isAcquiredInThisProcess());
    }
}

/**
 * Release a held lock on a dataset. There is no danger in calling this method if the lock is not held.
 *
 * @param datasetKey the dataset for which the held lock should be released
 */
public void releaseLock(String datasetKey) {
  String path = buildPath(datasetKey);
  InterProcessSemaphoreMutex lock = locks.get(path);
  try {
    if (lock != null) {
      lock.release();
    }
  } catch (Exception e) {
    LOG.warn("Failure communicating with Zookeeper", e);
  } finally {
    // if we fail to contact zookeeper we've already lost the lock
    locks.remove(path);
  }
}

protected InterProcessLock getLock(final IRI identifier) {
    final String path = ZNODE_COORDINATION + PATH_SEPARATOR + md5Hex(identifier.getIRIString());
    return new InterProcessSemaphoreMutex(curator, path);
}

private ZKLock(InterProcessSemaphoreMutex lock) {
  this.lock = lock;
}

public InterProcessSemaphoreMutex getLock() {
  return lock;
}

public InterProcessLock createLock(final String path) {
    return new InterProcessSemaphoreMutex(zkHolder.get(), path);
}

private DistributedLock(CuratorFramework client, String path) {
  this.path = path;
  this.lock = new InterProcessSemaphoreMutex(client, path);
}

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

/**
 * Process the given migration set
 *
 * @param set the set
 * @return completion stage. If there is a migration-specific error, the stage will be completed
 * exceptionally with {@link org.apache.curator.x.async.migrations.MigrationException}.
 */
public CompletionStage<Void> migrate(MigrationSet set)
{
    InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id()));
    CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor);
    return lockStage.thenCompose(__ -> runMigrationInLock(lock, set));
}