import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
private Completable initializeCurator()
{
    // Create/start CuratorFramework client
    RetryPolicy retryPolicy = new BoundedExponentialBackoffRetry(retryBaseTime, retryMaxTime, retryLimit);
    EnsembleProvider ensembleProvider = new FixedEnsembleProvider(connectionString);
    curator = CuratorFrameworkFactory.builder()
        .ensembleProvider(ensembleProvider)
        .retryPolicy(retryPolicy)
        .namespace(namespace)
        .sessionTimeoutMs(sessionTimeout)
        .connectionTimeoutMs(connectionTimeout)
        .build();
    curator.start();

    // Create a NodeCache for each config descriptor
    // This creates N node caches at a time on the RxJava IO scheduler thread pool.
    return Observable.from(configDescriptors)
        .flatMap(desc -> buildNodeCache(desc)
            .subscribeOn(Schedulers.io())
            .map(nc -> this.configNodeCaches.put(desc, nc)), getConcurrentNodeCacheCreations())
        .toCompletable();
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
@Before
public void init() throws Throwable {
    zkRootClient = CuratorFrameworkFactory.builder()
            .connectString(server.getConnectString())
            .retryPolicy(new BoundedExponentialBackoffRetry(10, 100, 7))
            .build();
    zkRootClient.start();

    ZooKeeper zk = zkRootClient.getZookeeperClient().getZooKeeper();
    ZKPaths.mkdirs(zk, "/"+ CloudConfigCommon.CONFIG_ROOT);
    ZKPaths.mkdirs(zk, "/"+CloudConfigCommon.PROPERTY_ROOT);

    zkConfigClient = zkRootClient.usingNamespace(CloudConfigCommon.CONFIG_ROOT);
    zkPropsClient  = zkRootClient.usingNamespace(CloudConfigCommon.PROPERTY_ROOT);

    prepare();
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
@Override
protected void before() throws Throwable
{
    ts = new TestingServer();
    curator = CuratorFrameworkFactory.builder()
                                     .namespace("ezkr")
                                     .connectString(ts.getConnectString())
                                     .retryPolicy(new BoundedExponentialBackoffRetry(10, 100, 7))
                                     .build();
    curator.getConnectionStateListenable().addListener(new ConnectionStateListener()
    {
        @Override
        public void stateChanged(final CuratorFramework client, final ConnectionState newState)
        {

        }
    });
    curator.start();
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public long nextId(final String namespace) {
    final String[] paths = calcPathIdAndPathLock(namespace);
    final String pathId = paths[0];
    final String pathLock = paths[1];

    RetryPolicy retryPolicyMutex = new BoundedExponentialBackoffRetry(10, 1000, 5);
    PromotedToLock promotedToLock = PromotedToLock.builder().retryPolicy(retryPolicyMutex)
            .lockPath(pathLock).build();
    RetryPolicy retryPolicyOptimistic = new RetryNTimes(3, 100);
    DistributedAtomicLong dal = new DistributedAtomicLong(curatorFramework, pathId,
            retryPolicyOptimistic, promotedToLock);
    semaphore.acquireUninterruptibly();
    try {
        AtomicValue<Long> value = dal.increment();
        if (value != null && value.succeeded()) {
            return value.postValue();
        }
        return -1;
    } catch (Exception e) {
        throw e instanceof IdException ? (IdException) e : new IdException(e);
    } finally {
        semaphore.release();
    }
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public long currentId(final String namespace) {
    final String[] paths = calcPathIdAndPathLock(namespace);
    final String pathId = paths[0];
    final String pathLock = paths[1];

    RetryPolicy retryPolicyMutex = new BoundedExponentialBackoffRetry(10, 1000, 5);
    PromotedToLock promotedToLock = PromotedToLock.builder().retryPolicy(retryPolicyMutex)
            .lockPath(pathLock).build();
    RetryPolicy retryPolicyOptimistic = new RetryNTimes(3, 100);
    DistributedAtomicLong dal = new DistributedAtomicLong(curatorFramework, pathId,
            retryPolicyOptimistic, promotedToLock);
    try {
        AtomicValue<Long> value = dal.get();
        if (value != null && value.succeeded()) {
            return value.postValue();
        }
        throw new IdException("Operation was not successful!");
    } catch (Exception e) {
        throw e instanceof IdException ? (IdException) e : new IdException(e);
    }
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
public RetryPolicy retryPolicy() {
    /**
     * int baseSleepTimeMs, int maxSleepTimeMs, int maxRetries
     **/
    int baseSleepTimeMs = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.base.sleep.time.ms", "1000"));
    int maxSleepTimeMs = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.max.sleep.time.ms", "5000"));
    int maxRetries = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.max.retries", "29"));
    return new BoundedExponentialBackoffRetry(baseSleepTimeMs,
            maxSleepTimeMs, maxRetries);
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
public RetryPolicy retryPolicy() {
	int baseSleepTimeMs = Integer.parseInt(env.getProperty(
			"rpc.server.zookeeper.base.sleep.time.ms", "1000"));
	int maxSleepTimeMs = Integer.parseInt(env.getProperty(
			"rpc.server.zookeeper.max.sleep.time.ms", "5000"));
	int maxRetries = Integer.parseInt(env.getProperty(
			"rpc.server.zookeeper.max.retries", "29"));
	return new BoundedExponentialBackoffRetry(baseSleepTimeMs,
			maxSleepTimeMs, maxRetries);
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
@Bean
public RetryPolicy retryPolicy() {
    int baseSleepTimeMs = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.base.sleep.time.ms", "1000"));
    int maxSleepTimeMs = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.max.sleep.time.ms", "5000"));
    int maxRetries = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.max.retries", "29"));
    return new BoundedExponentialBackoffRetry(baseSleepTimeMs,
            maxSleepTimeMs, maxRetries);
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
public static CuratorFramework getCuratorClient(final TrellisConfiguration config) {
    final CuratorFramework curator = newClient(config.getZookeeper().getEnsembleServers(),
            new BoundedExponentialBackoffRetry(config.getZookeeper().getRetryMs(),
                config.getZookeeper().getRetryMaxMs(), config.getZookeeper().getRetryMax()));
    curator.start();
    return curator;
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
public void init(Map<String, Object> conf, String participantId) {
    Preconditions.checkNotNull(participantId, "participantId can not be null");
    Preconditions.checkNotNull(conf, "conf can not be null");

    this.conf = conf;
    this.serverUrl = participantId;
    this.leaderLatchListener = createLeaderLatchListener();

    LOG.info("Received configuration : [{}]", conf);

    CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
    String url = (String) conf.get(CONNECT_URL);
    String rootPrefix = (String) conf.get("root");
    builder.connectString(url);
    builder.connectionTimeoutMs((Integer) conf.getOrDefault(CONNECTION_TIMEOUT_MS, DEFAULT_CONN_TIMOUT));
    builder.sessionTimeoutMs((Integer) conf.getOrDefault(SESSION_TIMEOUT_MS, DEFAULT_SESSION_TIMEOUT));

    builder.retryPolicy(
            new BoundedExponentialBackoffRetry(
                    (Integer) conf.getOrDefault(RETRY_BASE_SLEEP_TIME_MS, DEFAULT_BASE_SLEEP_TIME),
                    (Integer) conf.getOrDefault(RETRY_MAX_SLEEP_TIME_MS, DEFAULT_MAX_SLEEP_TIME),
                    (Integer) conf.getOrDefault(RETRY_LIMIT, DEFAULT_RETRY_LIMIT)

            ));

    curatorFramework = builder.build();
    leaderLatchPath = rootPrefix + LEADER_LOCK_NODE_PATH;
    leaderLatchRef = new AtomicReference<>(createLeaderLatch());
    curatorFramework.start();
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
@BeforeMethod
private void setup() throws Exception {
    _zooKeeperServer = new TestingServer();
    _curator = CuratorFrameworkFactory.newClient(_zooKeeperServer.getConnectString(),
            new BoundedExponentialBackoffRetry(100, 1000, 5));
    _curator.start();
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
public ZkRegistry(ZkConfig zkConfig) {

        zkClient = CuratorFrameworkFactory.builder().connectString(zkConfig.getZkAddress())
                .sessionTimeoutMs(zkConfig.getZkTimeout())
                .retryPolicy(new BoundedExponentialBackoffRetry(zkConfig.getBaseSleepTimeMs(),
                        zkConfig.getMaxSleepTimeMs(), zkConfig.getMaxRetries()))
                .build();

        zkClient.start();
    }
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
private static void prepare(TestingServer server) throws Exception {
        CuratorFramework zkRootClient = null;
        try {
            zkRootClient = CuratorFrameworkFactory.builder()
                    .connectString(server.getConnectString())
                    .retryPolicy(new BoundedExponentialBackoffRetry(10, 100, 7))
                    .build();
            zkRootClient.start();

            ZooKeeper zk = zkRootClient.getZookeeperClient().getZooKeeper();
            ZKPaths.mkdirs(zk, "/" + CloudConfigCommon.CONFIG_ROOT);
            ZKPaths.mkdirs(zk, "/"+CloudConfigCommon.PROPERTY_ROOT);

            CuratorFramework zkConfigClient = zkRootClient.usingNamespace(CloudConfigCommon.CONFIG_ROOT);
//        CuratorFramework zkPropsClient  = zkRootClient.usingNamespace(CloudConfigCommon.PROPERTY_ROOT);

            String config = "{\n" +
                    "    \"driverClassName\" : \"com.mysql.jdbc.Driver\",\n" +
                    "    \"userName\" : \"root\",\n" +
                    "    \"password\" : \"1111\", \n"+
                    "    \"jdbcUrl\" : \"jdbc:mysql://127.0.0.1:3306/a?characterEncoding=utf8&createDatabaseIfNotExist=true\"\n"+
                    "}";

            zkConfigClient.create().creatingParentsIfNeeded().forPath("/database/mydb", config.getBytes());
        } finally {
            if(zkRootClient!=null) {
                zkRootClient.close();
            }
        }
    }
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
private static Supplier<CuratorFramework> curatorSupplier() {
    return new Supplier<CuratorFramework>() {
        @Override public CuratorFramework get() {
            final String quorum = System.getProperty("zk.quorum", "localhost");
            final int sessionTimeout = Integer.parseInt(
                    System.getProperty("zk.session.timeout", "30000"));
            final int connectionTimeout = Integer.parseInt(
                    System.getProperty("zk.connection.timeout", "15000"));
            final int initialDelay = Integer.parseInt(
                    System.getProperty("zk.retry.initialDelay", "10"));
            final int maxDelay = Integer.parseInt(
                    System.getProperty("zk.retry.maxDelay", "200"));
            final int maxCount = Integer.parseInt(
                    System.getProperty("zk.retry.maxCount", "10"));


            logger.info("Initializing the Zookeeper client for quorum: {}", quorum);

            final CuratorFramework curator = CuratorFrameworkFactory.newClient(quorum, sessionTimeout, connectionTimeout,
                    new BoundedExponentialBackoffRetry(initialDelay, maxDelay, maxCount));

            curator.start();

            Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run() {
                    logger.info("Shutting down the Zookeeper client...");
                    curator.close();
                }
            });

            return curator;
        }
    };
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
@Override
RetryPolicy build(Config config) {
  return new BoundedExponentialBackoffRetry(
      getMillis(config, "baseSleepDuration"),
      getMillis(config, "maxSleepDuration"),
      config.getInt("maxRetries"));
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * 
 * @since 0.4.0
 */
@Override
public boolean setValue(String namespace, long value) {
    if (value < 0) {
        throw new IdException("Id value must be greater or equal to 0!");
    }

    final String[] paths = calcPathIdAndPathLock(namespace);
    final String pathId = paths[0];
    final String pathLock = paths[1];

    RetryPolicy retryPolicyMutex = new BoundedExponentialBackoffRetry(10, 1000, 5);
    PromotedToLock promotedToLock = PromotedToLock.builder().retryPolicy(retryPolicyMutex)
            .lockPath(pathLock).build();
    RetryPolicy retryPolicyOptimistic = new RetryNTimes(3, 100);
    DistributedAtomicLong dal = new DistributedAtomicLong(curatorFramework, pathId,
            retryPolicyOptimistic, promotedToLock);
    semaphore.acquireUninterruptibly();
    try {
        dal.forceSet(value);
        return true;
    } catch (Exception e) {
        throw e instanceof IdException ? (IdException) e : new IdException(e);
    } finally {
        semaphore.release();
    }
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
@Test
public void testDeserializeBoundedExponentialBackoffRetry() {
    ZooKeeperConfiguration config = parse(ImmutableMap.of("retryPolicy",
            ImmutableMap.builder()
                    .put("type", "boundedExponentialBackoff")
                    .put("baseSleepTimeMs", 50)
                    .put("maxSleepTimeMs", 500)
                    .put("maxRetries", 3)
                    .build()));
    assertTrue(config.getRetryPolicy().get() instanceof BoundedExponentialBackoffRetry);
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
private RetryPolicy retryPolicy() {
	return new BoundedExponentialBackoffRetry(baseSleepTimeMs,
			maxSleepTimeMs, maxRetries);
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
/**
 * Create a new curator instance off the root path; using configuration
 * options provided in the service configuration to set timeouts and
 * retry policy.
 * @return the newly created creator
 */
private CuratorFramework createCurator() throws IOException {
  Configuration conf = getConfig();
  createEnsembleProvider();
  int sessionTimeout = conf.getInt(KEY_REGISTRY_ZK_SESSION_TIMEOUT,
      DEFAULT_ZK_SESSION_TIMEOUT);
  int connectionTimeout = conf.getInt(KEY_REGISTRY_ZK_CONNECTION_TIMEOUT,
      DEFAULT_ZK_CONNECTION_TIMEOUT);
  int retryTimes = conf.getInt(KEY_REGISTRY_ZK_RETRY_TIMES,
      DEFAULT_ZK_RETRY_TIMES);
  int retryInterval = conf.getInt(KEY_REGISTRY_ZK_RETRY_INTERVAL,
      DEFAULT_ZK_RETRY_INTERVAL);
  int retryCeiling = conf.getInt(KEY_REGISTRY_ZK_RETRY_CEILING,
      DEFAULT_ZK_RETRY_CEILING);

  if (LOG.isDebugEnabled()) {
    LOG.debug("Creating CuratorService with connection {}",
        connectionDescription);
  }
  CuratorFramework framework;

  synchronized (CuratorService.class) {
    // set the security options

    // build up the curator itself
    CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
    builder.ensembleProvider(ensembleProvider)
     .connectionTimeoutMs(connectionTimeout)
     .sessionTimeoutMs(sessionTimeout)

     .retryPolicy(new BoundedExponentialBackoffRetry(retryInterval,
         retryCeiling,
         retryTimes));

    // set up the builder AND any JVM context
    registrySecurity.applySecurityEnvironment(builder);
    //log them
    securityConnectionDiagnostics = buildSecurityDiagnostics();
    framework = builder.build();
    framework.start();
  }

  return framework;
}
 

import org.apache.curator.retry.BoundedExponentialBackoffRetry; //导入依赖的package包/类
@Provides
@Singleton
CuratorFramework provideCuratorFramework(
    ShutdownRegistry shutdownRegistry,
    @ServiceDiscoveryBindings.ZooKeeper Iterable<InetSocketAddress> zooKeeperCluster,
    ACLProvider aclProvider) {

  String connectString =
      FluentIterable.from(zooKeeperCluster)
          .transform(InetSocketAddressHelper::toString)
          .join(Joiner.on(','));

  if (zooKeeperConfig.getChrootPath().isPresent()) {
    connectString = connectString + zooKeeperConfig.getChrootPath().get();
  }

  // This emulates the default BackoffHelper configuration used by the legacy commons/zookeeper
  // stack. BackoffHelper is unbounded, this dies after around 5 minutes using the 10 retries.
  // NB: BoundedExponentialBackoffRetry caps max retries at 29 if you send it a larger value.
  RetryPolicy retryPolicy =
      new BoundedExponentialBackoffRetry(
          Amount.of(1, Time.SECONDS).as(Time.MILLISECONDS),
          Amount.of(1, Time.MINUTES).as(Time.MILLISECONDS),
          10);

  CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
      .dontUseContainerParents() // Container nodes are only available in ZK 3.5+.
      .connectString(connectString)
      .canBeReadOnly(false) // We must be able to write to perform leader election.
      .sessionTimeoutMs(zooKeeperConfig.getSessionTimeout().as(Time.MILLISECONDS))
      .retryPolicy(retryPolicy)
      .aclProvider(aclProvider);

  if (zooKeeperConfig.getCredentials().isPresent()) {
    Credentials credentials = zooKeeperConfig.getCredentials().get();
    builder.authorization(credentials.scheme(), credentials.authToken());
  }

  CuratorFramework curatorFramework = builder.build();

  // TODO(John Sirois): It would be nice to use a Service to control the lifecycle here, but other
  // services (org.apache.aurora.scheduler.http.JettyServerModule.RedirectMonitor) rely on this
  // service being started 1st which is not deterministic as things stand.  Find a way to leverage
  // the Service system for services with Service dependencies.
  curatorFramework.start();
  shutdownRegistry.addAction(curatorFramework::close);

  return curatorFramework;
}