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

public CacheClusterViewer(Config config) throws RuntimeException {
    try {
        if (config == null) {
            config = ConfigFactory.load();
        }

        this.zookeeperConnectionUrl = config.getString("zookeeper.connection_url");

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        zkClient = CuratorFrameworkFactory.newClient(zookeeperConnectionUrl, retryPolicy);
        zkClient.start();

        clusterGlobalLock = new InterProcessReadWriteLock(zkClient, Constants.CACHE_CLUSTER_PATH);

        cacheCluster = doGetCacheClusterMeta();

        monitorThread = new Thread(new ClusterStatusMonitor());
        monitorThread.start();
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

private CacheClusterService(Config config) {
    try {
        if (config == null) {
            config = ConfigFactory.load();
        }

        this.zookeeperConnectionUrl = config.getString("zookeeper.connection_url");

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        zkClient = CuratorFrameworkFactory.newClient(zookeeperConnectionUrl, retryPolicy);
        zkClient.start();

        clusterGlobalLock = new InterProcessReadWriteLock(zkClient, Constants.CACHE_CLUSTER_PATH);

        cacheClusterViewer = CacheClusterViewerFactory.getCacheClusterViewer();
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

public ZkConfig(String connectStr,String baseDir,boolean readonly){
    this.readonly = readonly;
    client = CuratorFrameworkFactory.builder().connectString(connectStr)
            .namespace(baseDir).retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))
            .connectionTimeoutMs(5000).build();
    client.start();
    try {
        client.blockUntilConnected();
        nameslock = new ZkDistributedRwLock( new InterProcessReadWriteLock(client, STORE_NAMES_Lock_PATH));
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
 * this function will lock the cache cluster to guarantee the integrity of CacheClusterMeta
 * in CacheClusterViewer, so must not call it during the cluster is locked.
 * <p>
 * on cache cluster servers, this function should be called before starting cache services to
 * make sure the deadlock dose not happen.
 * <p>
 * on cache cluster clients, it's recommend this method called only once before getInstance
 * called, then all calls of getInstance will return the same CacheClusterViewer instance.
 */
public static void configure(Config config) throws Exception {
    if (config == null) {
        config = ConfigFactory.load();
    }

    String zookeeperConnectionUrl = config.getString("zookeeper.connection_url");
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    CuratorFramework zkClient = CuratorFrameworkFactory.newClient(zookeeperConnectionUrl, retryPolicy);
    zkClient.start();

    InterProcessReadWriteLock clusterGlobalLock =
            new InterProcessReadWriteLock(zkClient, Constants.CACHE_CLUSTER_PATH);
    clusterGlobalLock.readLock().acquire();
    try {
        cacheClusterViewer = new CacheClusterViewer(config);
    } finally {
        try {
            clusterGlobalLock.readLock().release();
        } catch (Exception e) {
            logger.error(String.format("failed to release clusterGlobalLock on zknode[%s]",
                    Constants.CACHE_CLUSTER_PATH), e);
        }
    }

    zkClient.close();
}

/**
 * This establishes the connection to zookeeper so that we can look up services.
 */
public ZooKeeperServiceRegistry(String zookeeperConnectString) {
    // We don't use .namespace() since we need the data stored in its own
    // path to maintain backwards-compatibility. We fake namespaces on our
    // own in formatZooKeeperApplicationPath.
    client = CuratorFrameworkFactory.builder()
            .connectString(zookeeperConnectString)
            .retryPolicy(new RetryNTimes(DEFAULT_MAX_NUM_OF_TRIES, DEFAULT_TIMEOUT))
            .build();
    client.start();

    readWriteLock = new InterProcessReadWriteLock(client, GLOBAL_LOCK_PATH);
}

public ExampleClientReadWriteLocks(CuratorFramework client, String lockPath, FakeLimitedResource resource, String clientName) {
    this.resource = resource;
    this.clientName = clientName;
    lock = new InterProcessReadWriteLock(client, lockPath);
    readLock = lock.readLock();
    writeLock = lock.writeLock();
}

public ZkDistributedRwLock(InterProcessReadWriteLock readWriteLock){
   this.readWriteLock = readWriteLock;
}

private InternalReadWriteLock(InterProcessReadWriteLock delegate) {
    this.delegate = delegate;
}