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

@Override
public void registerInstance(ServiceInstance instance) throws DiscoveryException {
    String path = formatZooKeeperEndpointPath(instance);
    InterProcessLock writeLock = readWriteLock.writeLock();
    Exception error = null;

    try {
        writeLock.acquire();

        if (client.checkExists().forPath(path) == null) {
            client.create().creatingParentsIfNeeded().forPath(path);
        }
    } catch(Exception e) {
        error = e;
    } finally {
        safelyRelease(writeLock, error);
    }
}

/**
 * Return a list of the children of a path or empty if the path does not exist.
 *
 * @param path ZooKeeper path to return children of
 * @return list of children of the path or empty if the path does not exist
 * @throws DiscoveryException on ZooKeeper error
 */
private List<String> childrenOrEmpty(String path) throws DiscoveryException {
    List<String> children = null;
    InterProcessLock readLock = readWriteLock.readLock();
    Exception error = null;

    try {
        readLock.acquire();

        if (client.checkExists().forPath(path) == null) {
            children = Collections.emptyList();
        } else {
            children = client.getChildren().forPath(path);
        }
    } catch (Exception e) {
        error = e;
    } finally {
        safelyRelease(readLock, error);
    }

    return children;
}

public void supervene() {
    final CountDownLatch countDownLatch = new CountDownLatch(1);
    final InterProcessLock interProcessLock = new InterProcessMutex(curatorFramework, "/lock");

    int count = 10;
    while (count > 0) {

        new Thread() {
            @Override
            public void run() {
                try {
                    countDownLatch.await();
                    interProcessLock.acquire();
                    String now = simpleDateFormat.format(new Date());
                    LOG.info("Now time: ".concat(now));
                    interProcessLock.release();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }.start();
        count--;
    }
    countDownLatch.countDown();
}

@Override
protected InterProcessLock getLock(final IRI identifier) {
    if (nonNull(lock)) {
        return lock;
    }
    return super.getLock(identifier);
}

@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));
}

public static <V> V runLocked(final Callable<V> callable, final InterProcessLock lock) throws Exception {
    lock.acquire();
    try {
        return callable.call();
    } finally {
        try {
            lock.release();
        } catch (final Exception e) {
            LOG.warn("Error occurred when releasing ZK lock", e);
        }
    }
}

@Override
public List<ServiceInstance> listInstances(String applicationName, String serviceName, String serviceType)
        throws DiscoveryException {
    String path = formatZooKeeperServiceTypePath(applicationName, serviceName, serviceType);
    List<ServiceInstance> instances = new ArrayList<ServiceInstance>();
    InterProcessLock readLock = readWriteLock.readLock();
    Exception maybeError = null;

    try {
        List<String> children;
        readLock.acquire();

        if (client.checkExists().forPath(path) == null) {
            children = Collections.emptyList();
        } else {
            children = client.getChildren().forPath(path);
        }

        for (String c : children) {
            HostAndPort hostAndPort = HostAndPort.fromString(c);
            instances.add(new BasicServiceInstance(applicationName, serviceName, serviceType,
                    hostAndPort.getHostText(), hostAndPort.getPort()));
        }

    } catch (Exception e) {
        maybeError = e;
    } finally {
        safelyRelease(readLock, maybeError);
    }

    return instances;
}

/**
 * Allow a lock to be released within a finally block.
 *
 * If parentError is non-null, it is thrown so that the original cause of
 * the exception can be reported back to the caller rather than hiding it
 * behind a potential error when releasing the lock.
 *
 * @param lock lock to release
 * @param parentError the exception causing us to enter the finally block or null if none exists
 * @throws DiscoveryException if the lock cannot be released
 */
private void safelyRelease(InterProcessLock lock, Exception parentError) throws DiscoveryException {
    try {
        lock.release();
    } catch (Exception e) {
        if (parentError != null) {
            throw new DiscoveryException(parentError);
        } else {
            throw new DiscoveryException(e);
        }
    }
}

private CompletionStage<Void> runMigrationInLock(InterProcessLock lock, MigrationSet set)
{
    String thisMetaDataPath = ZKPaths.makePath(metaDataPath, set.id());
    return childrenWithData(client, thisMetaDataPath)
        .thenCompose(metaData -> applyMetaData(set, metaData, thisMetaDataPath))
        .handle((v, e) -> {
            release(lock, true);
            if ( e != null )
            {
                Throwables.propagate(e);
            }
            return v;
        }
    );
}

/**
 * Attempt to acquire the given lock asynchronously using the given timeout and executor. If the lock
 * is not acquired within the timeout stage is completedExceptionally with {@link AsyncWrappers.TimeoutException}
 *
 * @param lock a lock implementation (e.g. {@link org.apache.curator.framework.recipes.locks.InterProcessMutex},
 * {@link org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2}, etc.)
 * @param timeout max timeout to acquire lock
 * @param unit time unit of timeout
 * @param executor executor to use to asynchronously acquire
 * @return stage
 */
public static CompletionStage<Void> lockAsync(InterProcessLock lock, long timeout, TimeUnit unit, Executor executor)
{
    CompletableFuture<Void> future = new CompletableFuture<>();
    if ( executor == null )
    {
        CompletableFuture.runAsync(() -> lock(future, lock, timeout, unit));
    }
    else
    {
        CompletableFuture.runAsync(() -> lock(future, lock, timeout, unit), executor);
    }
    return future;
}

/**
 * Attempt to acquire the given lock asynchronously using the given timeout and executor. The stage
 * is completed with a Boolean that indicates whether or not the lock was acquired.
 *
 * @param lock a lock implementation (e.g. {@link org.apache.curator.framework.recipes.locks.InterProcessMutex},
 * {@link org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2}, etc.)
 * @param timeout max timeout to acquire lock
 * @param unit time unit of timeout
 * @param executor executor to use to asynchronously acquire
 * @return stage
 */
public static CompletionStage<Boolean> lockAsyncIf(InterProcessLock lock, long timeout, TimeUnit unit, Executor executor)
{
    CompletableFuture<Boolean> future = new CompletableFuture<>();
    if ( executor == null )
    {
        CompletableFuture.runAsync(() -> lockIf(future, lock, timeout, unit));
    }
    else
    {
        CompletableFuture.runAsync(() -> lockIf(future, lock, timeout, unit), executor);
    }
    return future;
}

private static void lockIf(CompletableFuture<Boolean> future, InterProcessLock lock, long timeout, TimeUnit unit)
{
    try
    {
        future.complete(lock.acquire(timeout, unit));
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        future.completeExceptionally(e);
    }
}

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());
    }
}

@Test
public void testMutexCall() throws Exception
{
    final ConfigurationManager manager  = new ConfigurationManager(client, reference, serializer, path);
    final InterProcessLock mutex        = mock(InterProcessLock.class);
    final Throwable throwable           = new RuntimeException();
    final Callable callable             = mock(Callable.class);

    manager.mutex(callable, mutex);

    verify(mutex).acquire();
    verify(callable).call();
    verify(mutex).release();

    reset(mutex, callable);

    when(callable.call()).thenThrow(throwable);

    try {
        manager.mutex(callable, mutex);
    } catch (Exception e) {
        assertSame(e, throwable);
    }

    verify(mutex).acquire();
    verify(callable).call();
    verify(mutex).release();
}

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

public MyResourceService(final String connectString, final EventService eventService,
        final InterProcessLock lock) {
    this(getZkClient(connectString), eventService, lock);
}

public MyResourceService(final CuratorFramework curator, final EventService eventService,
        final InterProcessLock lock) {
    super(baseUrl, new MockProducer<>(true, new StringSerializer(), new StringSerializer()), curator,
            eventService, mockIdSupplier, false);
    this.lock = lock;
}

/** Create a mutex which ensures exclusive access within this single vm */
@Override
public InterProcessLock createMutex(String path) {
    return new MockLock(path);
}

/** For internal use; prefer creating a {@link com.yahoo.vespa.curator.recipes.CuratorLock} */
public InterProcessLock createMutex(String lockPath) {
    return new InterProcessMutex(curatorFramework, lockPath);
}

public ZookeeperPseudoLock(InterProcessLock lock)
{
    this.lock = lock;
}

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

/**
 * 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));
}

/**
 * Attempt to acquire the given lock asynchronously using the given executor and without a timeout.
 *
 * @param lock a lock implementation (e.g. {@link org.apache.curator.framework.recipes.locks.InterProcessMutex},
 * {@link org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2}, etc.)
 * @param executor executor to use to asynchronously acquire
 * @return stage
 */
public static CompletionStage<Void> lockAsync(InterProcessLock lock, Executor executor)
{
    return lockAsync(lock, 0, null, executor);
}

/**
 * Attempt to acquire the given lock asynchronously using the given timeout using the {@link java.util.concurrent.ForkJoinPool#commonPool()}.
 * If the lock is not acquired within the timeout stage is completedExceptionally with {@link AsyncWrappers.TimeoutException}
 *
 * @param lock a lock implementation (e.g. {@link org.apache.curator.framework.recipes.locks.InterProcessMutex},
 * {@link org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2}, etc.)
 * @param timeout max timeout to acquire lock
 * @param unit time unit of timeout
 * @return stage
 */
public static CompletionStage<Void> lockAsync(InterProcessLock lock, long timeout, TimeUnit unit)
{
    return lockAsync(lock, timeout, unit, null);
}

/**
 * Attempt to acquire the given lock asynchronously using the given timeout using the {@link java.util.concurrent.ForkJoinPool#commonPool()}.
 * The stage is completed with a Boolean that indicates whether or not the lock was acquired.
 *
 * @param lock a lock implementation (e.g. {@link org.apache.curator.framework.recipes.locks.InterProcessMutex},
 * {@link org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2}, etc.)
 * @param timeout max timeout to acquire lock
 * @param unit time unit of timeout
 * @return stage
 */
public static CompletionStage<Boolean> lockAsyncIf(InterProcessLock lock, long timeout, TimeUnit unit)
{
    return lockAsyncIf(lock, timeout, unit, null);
}

/**
 * Attempt to acquire the given lock asynchronously without timeout using the {@link java.util.concurrent.ForkJoinPool#commonPool()}.
 *
 * @param lock a lock implementation (e.g. {@link org.apache.curator.framework.recipes.locks.InterProcessMutex},
 * {@link org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2}, etc.)
 * @return stage
 */
public static CompletionStage<Void> lockAsync(InterProcessLock lock)
{
    return lockAsync(lock, 0, null, null);
}

/**
 * Release the lock and wrap any exception in <code>RuntimeException</code>
 *
 * @param lock lock to release
 */
public static void release(InterProcessLock lock)
{
    release(lock, true);
}