Java 类org.apache.curator.framework.CuratorFrameworkFactory 实例源码

项目：fastmq 文件：LogStorageImplTest.java

@Before
public void setUp() throws Exception {
    CountDownLatch latch = new CountDownLatch(1);
    zookeeper = new ZooKeeper("127.0.0.1:2181", 10000, event -> {
        if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
            System.out.println("Zookeeper connected.");
        } else {
            throw new RuntimeException("Error connecting to zookeeper");
        }
        latch.countDown();
    });
    latch.await();
    CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("127.0.0.1:2181", new ExponentialBackoffRetry(1000, 3));
    curatorFramework.start();
    AsyncCuratorFramework asyncCuratorFramework = AsyncCuratorFramework.wrap(curatorFramework);
    logInfoStorage = new LogInfoStorageImpl(asyncCuratorFramework);
}

项目：iotplatform 文件：ZkDiscoveryService.java

@PostConstruct
public void init() {
  log.info("Initializing...");
  Assert.hasLength(zkUrl, MiscUtils.missingProperty("zk.url"));
  Assert.notNull(zkRetryInterval, MiscUtils.missingProperty("zk.retry_interval_ms"));
  Assert.notNull(zkConnectionTimeout, MiscUtils.missingProperty("zk.connection_timeout_ms"));
  Assert.notNull(zkSessionTimeout, MiscUtils.missingProperty("zk.session_timeout_ms"));

  log.info("Initializing discovery service using ZK connect string: {}", zkUrl);

  zkNodesDir = zkDir + "/nodes";
  try {
    client = CuratorFrameworkFactory.newClient(zkUrl, zkSessionTimeout, zkConnectionTimeout,
        new RetryForever(zkRetryInterval));
    client.start();
    client.blockUntilConnected();
    cache = new PathChildrenCache(client, zkNodesDir, true);
    cache.getListenable().addListener(this);
    cache.start();
  } catch (Exception e) {
    log.error("Failed to connect to ZK: {}", e.getMessage(), e);
    CloseableUtils.closeQuietly(client);
    throw new RuntimeException(e);
  }
}

项目：Mastering-Mesos 文件：SingularityCuratorProvider.java

@Inject
public SingularityCuratorProvider(final SingularityConfiguration configuration, final Set<ConnectionStateListener> connectionStateListeners) {

  checkNotNull(configuration, "configuration is null");
  checkNotNull(connectionStateListeners, "connectionStateListeners is null");

  ZooKeeperConfiguration zookeeperConfig = configuration.getZooKeeperConfiguration();

  this.curatorFramework = CuratorFrameworkFactory.builder()
      .defaultData(null)
      .sessionTimeoutMs(zookeeperConfig.getSessionTimeoutMillis())
      .connectionTimeoutMs(zookeeperConfig.getConnectTimeoutMillis())
      .connectString(zookeeperConfig.getQuorum())
      .retryPolicy(new ExponentialBackoffRetry(zookeeperConfig.getRetryBaseSleepTimeMilliseconds(), zookeeperConfig.getRetryMaxTries()))
      .namespace(zookeeperConfig.getZkNamespace()).build();

  for (ConnectionStateListener connectionStateListener : connectionStateListeners) {
    curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
  }
}

项目：simpleJobScheduler 文件：Application.java

@Override
public void run(String... strings) throws Exception {
    client = CuratorFrameworkFactory.newClient(zookeeperConnString,
            new ExponentialBackoffRetry(1000, Integer.MAX_VALUE));
    client.start();
    client.getZookeeperClient().blockUntilConnectedOrTimedOut();

    leaderLatch = new LeaderLatch(client, "/http-job-scheduler/leader",
            ManagementFactory.getRuntimeMXBean().getName());

    leaderLatch.addListener(new LeaderLatchListener() {
        @Override
        public void isLeader() {
            setMaster(true);
            masterJobScheduler.resume();
        }

        @Override
        public void notLeader() {
            setMaster(false);
            masterJobScheduler.pause();
        }
    });
    leaderLatch.start();
}

项目：fastmq 文件：ZkOffsetStorageImplTest.java

@Before
public void setUp() throws Exception {
    Configurator
        .initialize("FastMQ", Thread.currentThread().getContextClassLoader(), "log4j2.xml");
    Log log = new Log();
    LogSegment logSegment = new LogSegment();
    logSegment.setLedgerId(ledgerId);
    logSegment.setTimestamp(System.currentTimeMillis());
    log.setSegments(Collections.singletonList(logSegment));
    when(logInfoStorage.getLogInfo(any())).thenReturn(log);

    CuratorFramework curatorFramework = CuratorFrameworkFactory
        .newClient("127.0.0.1:2181", new ExponentialBackoffRetry(1000, 3));
    curatorFramework.start();
    asyncCuratorFramework = AsyncCuratorFramework.wrap(curatorFramework);
    offsetStorage = new ZkOffsetStorageImpl(logInfoStorage, asyncCuratorFramework);
}

项目：Juice 文件：CuratorUtils.java

public CuratorUtils(String connectString, Host host) {
    this.host = host;

    client = CuratorFrameworkFactory.builder()
            .connectString(connectString)
            .sessionTimeoutMs(10000)
            .connectionTimeoutMs(10000)
            .retryPolicy(new ExponentialBackoffRetry(1000, 3))
            .build();
    client.start();
    try {
        path = generatePath();
        if (null == path) {
            log.error("init curator failed due to path is null!");
            throw new DriverException("init curator failed due to path is null, servie will down");
        }
    } catch (Exception e) {
        log.error("CuratorUtils construct failed, service will down!");
        client.close();
        System.exit(-1);
    }
}

项目：jigsaw-payment 文件：HelloClientConfig.java

@Bean(initMethod = "start", destroyMethod = "close")
@Lazy(false)
public CuratorFramework curatorFramework() {
    CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory
            .builder();
    int sessionTimeoutMs = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.session.timeout.ms", "5000"));
    int connectionTimeoutMs = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.connection.timeout.ms", "5000"));
    builder.connectString(
            env.getProperty("rpc.client.zookeeper.connect.string"))
            .sessionTimeoutMs(sessionTimeoutMs)
            .connectionTimeoutMs(connectionTimeoutMs)

            .retryPolicy(this.retryPolicy());
            //.aclProvider(this.aclProvider()).authorization(this.authInfo());
    return builder.build();
}

项目：jigsaw-payment 文件：HelloServerConfig.java

@Bean(name = "curator-framework")
public CuratorFramework curatorFramework() {
    return CuratorFrameworkFactory
            .builder()
            .connectString(
                    env.getProperty("rpc.server.zookeeper.connect.string"))
            .sessionTimeoutMs(
                    Integer.parseInt(env.getProperty(
                            "rpc.server.zookeeper.session.timeout.ms",
                            "10000")))
            .connectionTimeoutMs(
                    Integer.parseInt(env.getProperty(
                            "rpc.server.zookeeper.connection.timeout.ms",
                            "10000"))).retryPolicy(this.retryPolicy())
            .aclProvider(this.aclProvider()).authorization(this.authInfo())
            .build();
}

项目：jigsaw-payment 文件：HelloClientConfig.java

@Bean(initMethod = "start", destroyMethod = "close")
public CuratorFramework curatorFramework() {
    CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory
            .builder();
    int sessionTimeoutMs = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.session.timeout.ms", "5000"));
    int connectionTimeoutMs = Integer.parseInt(env.getProperty(
            "rpc.client.zookeeper.connection.timeout.ms", "5000"));
    builder.connectString(
            env.getProperty("rpc.client.zookeeper.connect.string"))
            .sessionTimeoutMs(sessionTimeoutMs)
            .connectionTimeoutMs(connectionTimeoutMs)                  
            .retryPolicy(this.retryPolicy())
            .aclProvider(this.aclProvider()).authorization(this.authInfo());
    return builder.build();
}

项目：DBus 文件：ZkService.java

/**
 * 创建ZK连接
 * @param connectString  ZK服务器地址列表
 * @param sessionTimeout   Session超时时间
 */
public ZkService(String connectString, int sessionTimeout) throws Exception {
    CuratorFrameworkFactory.Builder builder;
    builder = CuratorFrameworkFactory.builder()
            .connectString(connectString)
            .namespace("")
            .authorization("digest", auth.getBytes())
            .retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))
            .connectionTimeoutMs(sessionTimeout);

    client = builder.build();
    client.start();
    if(!client.blockUntilConnected(20, TimeUnit.SECONDS)) {
        throw new Exception("zookeeper connected failed!");
    }

    tableVersions = new HashMap<>();
    cache = new HashMap<>();
}

项目：stroom-stats 文件：ServiceDiscoveryCuratorFrameworkProvider.java

@Override
public CuratorFramework get() {
    String quorum = zookeeperConfig.getQuorum();
    String serviceDiscoveryPath = zookeeperConfig.getServiceDiscoveryPath();
    String connectionString = quorum + (serviceDiscoveryPath == null ? "" : serviceDiscoveryPath);

    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

    LOGGER.info("Initiating Curator connection to Zookeeper using: [{}]", connectionString);
    // Use chroot so all subsequent paths are below /stroom-services to avoid conflicts with hbase/zookeeper/kafka etc.
    CuratorFramework client = CuratorFrameworkFactory.newClient(connectionString, retryPolicy);
    client.start();

    try {
        //Ensure the chrooted path for stroom-services exists
        Stat stat = client.checkExists().forPath("/");
        if (stat == null) {
            LOGGER.info("Creating chroot-ed root node inside " + serviceDiscoveryPath);
            client.create().forPath("/");
        }
    } catch (Exception e) {
        throw new RuntimeException("Error connecting to zookeeper using connection String: " + connectionString, e);
    }
    return client;
}

项目：stroom-stats 文件：StroomPropertyServiceCuratorFrameworkProvider.java

@Override
public CuratorFramework get() {
    String quorum = zookeeperConfig.getQuorum();
    String statsPath = zookeeperConfig.getPropertyServicePath();
    String connectionString = quorum + (statsPath == null ? "" : statsPath);

    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

    LOGGER.info("Initiating Curator connection to Zookeeper using: [{}]", connectionString);
    // Use chroot so all subsequent paths are below /stroom-stats to avoid conflicts with hbase/zookeeper/kafka etc.
    CuratorFramework client = CuratorFrameworkFactory.newClient(connectionString, retryPolicy);
    client.start();

    try {
        // Ensure the chrooted root path exists (i.e. /stroom-stats)
        Stat stat = client.checkExists().forPath("/");
        if (stat == null) {
            LOGGER.info("Creating chroot-ed root node inside " + statsPath);
            client.create().forPath("/");
        }
    } catch (Exception e) {
        throw new RuntimeException("Error connecting to zookeeper using connection String: " + connectionString, e);
    }
    return client;
}

项目：mycat-src-1.6.1-RELEASE 文件：ZktoXmlMain.java

private static CuratorFramework buildConnection(String url) {
    CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(url, new ExponentialBackoffRetry(100, 6));

    // start connection
    curatorFramework.start();
    // wait 3 second to establish connect
    try {
        curatorFramework.blockUntilConnected(3, TimeUnit.SECONDS);
        if (curatorFramework.getZookeeperClient().isConnected()) {
            return curatorFramework.usingNamespace("");
        }
    } catch (InterruptedException ignored) {
        Thread.currentThread().interrupt();
    }

    // fail situation
    curatorFramework.close();
    throw new RuntimeException("failed to connect to zookeeper service : " + url);
}

项目：mycat-src-1.6.1-RELEASE 文件：XmltoZkMain.java

private static CuratorFramework buildConnection(String url) {
    CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(url, new ExponentialBackoffRetry(100, 6));

    // start connection
    curatorFramework.start();
    // wait 3 second to establish connect
    try {
        curatorFramework.blockUntilConnected(3, TimeUnit.SECONDS);
        if (curatorFramework.getZookeeperClient().isConnected()) {
            return curatorFramework.usingNamespace("");
        }
    } catch (InterruptedException ignored) {
        Thread.currentThread().interrupt();
    }

    // fail situation
    curatorFramework.close();
    throw new RuntimeException("failed to connect to zookeeper service : " + url);
}

项目：mycat-src-1.6.1-RELEASE 文件：ZKUtils.java

private static CuratorFramework createConnection() {
       String url= ZkConfig.getInstance().getZkURL();

    CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(url, new ExponentialBackoffRetry(100, 6));

    // start connection
    curatorFramework.start();
    // wait 3 second to establish connect
    try {
        curatorFramework.blockUntilConnected(3, TimeUnit.SECONDS);
        if (curatorFramework.getZookeeperClient().isConnected()) {
            return curatorFramework;
        }
    } catch (InterruptedException ignored) {
        Thread.currentThread().interrupt();
    }

    // fail situation
    curatorFramework.close();
    throw new RuntimeException("failed to connect to zookeeper service : " + url);
}

项目：elastic-job-cloud 文件：ZookeeperRegistryCenterModifyTest.java

@Test
public void assertPersistEphemeralSequential() throws Exception {
    zkRegCenter.persistEphemeralSequential("/sequential/test_ephemeral_sequential");
    zkRegCenter.persistEphemeralSequential("/sequential/test_ephemeral_sequential");
    CuratorFramework client = CuratorFrameworkFactory.newClient(EmbedTestingServer.getConnectionString(), new RetryOneTime(2000));
    client.start();
    client.blockUntilConnected();
    List<String> actual = client.getChildren().forPath("/" + ZookeeperRegistryCenterModifyTest.class.getName() + "/sequential");
    assertThat(actual.size(), is(2));
    for (String each : actual) {
        assertThat(each, startsWith("test_ephemeral_sequential"));
    }
    zkRegCenter.close();
    actual = client.getChildren().forPath("/" + ZookeeperRegistryCenterModifyTest.class.getName() + "/sequential");
    assertTrue(actual.isEmpty());
    zkRegCenter.init();
}

项目：elastic-job-cloud 文件：ZookeeperElectionServiceTest.java

@Test
@Ignore
public void assertContend() throws Exception {
    CuratorFramework client = CuratorFrameworkFactory.newClient(EmbedTestingServer.getConnectionString(), new RetryOneTime(2000));
    client.start();
    client.blockUntilConnected();
    ZookeeperElectionService service = new ZookeeperElectionService(HOST_AND_PORT, client, ELECTION_PATH, electionCandidate);
    service.start();
    ElectionCandidate anotherElectionCandidate = mock(ElectionCandidate.class);
    CuratorFramework anotherClient = CuratorFrameworkFactory.newClient(EmbedTestingServer.getConnectionString(), new RetryOneTime(2000));
    ZookeeperElectionService anotherService = new ZookeeperElectionService("ANOTHER_CLIENT:8899", anotherClient, ELECTION_PATH, anotherElectionCandidate);
    anotherClient.start();
    anotherClient.blockUntilConnected();
    anotherService.start();
    KillSession.kill(client.getZookeeperClient().getZooKeeper(), EmbedTestingServer.getConnectionString());
    service.stop();
    verify(anotherElectionCandidate).startLeadership();
}

项目：Cobweb 文件：SelfTest.java

/**
 * 检查ZooKeeper的连接状态和它的Znode目录树
 *
 * @param
 * @return
 */
public static CuratorFramework checkAndGetZK() {
    CuratorFramework client;
    try {
        RetryPolicy retryPolicy =
                new ExponentialBackoffRetry(configuration.ZK_RETRY_INTERVAL, configuration.ZK_RETRY_TIMES);
        client = CuratorFrameworkFactory
                .newClient(configuration.ZK_CONNECT_STRING
                        , configuration.ZK_SESSION_TIMEOUT, configuration.ZK_INIT_TIMEOUT, retryPolicy);
        client.start();
        client.checkExists().forPath(ZNodeStaticSetting.TASKS_PATH);
        client.checkExists().forPath(ZNodeStaticSetting.MANAGERS_PATH);
        client.checkExists().forPath(ZNodeStaticSetting.WORKERS_PATH);
        client.checkExists().forPath(ZNodeStaticSetting.FILTERS_ROOT);
    } catch (Throwable e) {
        client = null;
        logger.error(e.getMessage());
    }
    return client;
}

项目：QDrill 文件：TestPStoreProviders.java

@Test
public void verifyZkStore() throws Exception {
  DrillConfig config = getConfig();
  String connect = config.getString(ExecConstants.ZK_CONNECTION);
  CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
  .namespace(config.getString(ExecConstants.ZK_ROOT))
  .retryPolicy(new RetryNTimes(1, 100))
  .connectionTimeoutMs(config.getInt(ExecConstants.ZK_TIMEOUT))
  .connectString(connect);

  try(CuratorFramework curator = builder.build()){
    curator.start();
    ZkPStoreProvider provider = new ZkPStoreProvider(config, curator);
    PStoreTestUtil.test(provider);
  }
}

项目：hadoop 文件：RegistrySecurity.java

/**
 * Apply the security environment to this curator instance. This
 * may include setting up the ZK system properties for SASL
 * @param builder curator builder
 */
public void applySecurityEnvironment(CuratorFrameworkFactory.Builder builder) {

  if (isSecureRegistry()) {
    switch (access) {
      case anon:
        clearZKSaslClientProperties();
        break;

      case digest:
        // no SASL
        clearZKSaslClientProperties();
        builder.authorization(SCHEME_DIGEST, digestAuthData);
        break;

      case sasl:
        // bind to the current identity and context within the JAAS file
        setZKSaslClientProperties(jaasClientIdentity, jaasClientContext);
    }
  }
}

项目：redirector 文件：Connector.java

private void createConnector() {
    CuratorFramework curator = CuratorFrameworkFactory.builder()
            .connectionTimeoutMs(specificConfig.getConnectionTimeoutMs())
            .retryPolicy(new RetryNTimes(specificConfig.getRetryCount(), specificConfig.getSleepsBetweenRetryMs()))
            .connectString(specificConfig.getConnectionUrl())
            .compressionProvider(new GzipCompressionCustomProvider())
            .build();

    setClient(curator);
    setBasePath(specificConfig.getZookeeperBasePath());
    setCacheHosts(specificConfig.isCacheHosts());
    setStacksCacheFactory(() -> new ZKStacksCache(curator, this, new ServiceDiscoveryHostJsonSerializer(),
            specificConfig.isCacheHosts(), specificConfig.getZookeeperBasePath()));

    setNodeCacheFactory(new ZkNodeCacheFactory(this, curator));
    setPathChildrenCacheFactory(new ZkPathChildrenCacheFactory(this, curator));
}

项目：redirector 文件：AbstractCommonBeans.java

@Bean
public CuratorFramework curatorFramework() {
    ZKConfig config = config();

    if (config.useZooKeeperWaitTimePolicy()) {
        return new RedirectorCuratorFramework(config);
    }

    CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
            .connectString(config.getZooKeeperConnection())
            .connectionTimeoutMs(config.getZooKeeperConnectionTimeout())
            .sessionTimeoutMs(config.getZooKeeperSessionTimeout())
            .retryPolicy(new RetryNTimes(config.getZooKeeperRetryAttempts(), config.getZooKeeperRetryInterval()))
            .compressionProvider(new GzipCompressionProvider());

    return builder.build();
}

项目：mumu-zookeeper 文件：CuratorServer.java

public void testServer(){
        try {
        TestingServer server=new TestingServer(2181,new File("/"));
        server.start();

        CuratorFramework curatorFramework = CuratorFrameworkFactory.
                builder().
                connectString(server.getConnectString()).
                sessionTimeoutMs(1000).
                retryPolicy(new RetryNTimes(3, 1000)).
                build();
        curatorFramework.start();
        System.out.println(curatorFramework.getChildren().forPath("/"));
        curatorFramework.close();
        server.stop();
    } catch (Exception e) {
        e.printStackTrace();
        System.exit(1);
    }
}

项目：dble 文件：ZKUtils.java

private static CuratorFramework createConnection() {
    String url = ZkConfig.getInstance().getZkURL();
    CuratorFramework framework = CuratorFrameworkFactory.newClient(url, new ExponentialBackoffRetry(100, 6));
    // start connection
    framework.start();
    // wait 3 second to establish connect
    try {
        framework.blockUntilConnected(3, TimeUnit.SECONDS);
        if (framework.getZookeeperClient().isConnected()) {
            LOGGER.info("CuratorFramework createConnection success");
            return framework;
        }
    } catch (InterruptedException ignored) {
        LOGGER.info("CuratorFramework createConnection error", ignored);
        Thread.currentThread().interrupt();
    }
    // fail situation
    framework.close();
    throw new RuntimeException("failed to connect to zookeeper service : " + url);
}

项目：albedo-thrift 文件：CommonAutoConfiguration.java

@Bean(destroyMethod = "close")
@ConditionalOnMissingBean
public CuratorFramework curatorFramework(AlbedoRpcProperties albedoRpcProperties, RetryPolicy retryPolicy) throws InterruptedException {
    CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
    String namespace = environment.getProperty("albedo.rpc.namespace", albedoRpcProperties.getNamespace());
    AlbedoRpcProperties.Zookeeper zookeeper = albedoRpcProperties.getZookeeper();
    String connectString = environment.getProperty("albedo.rpc.zookeeper.connectString", zookeeper.getConnectString());
    Integer blockUntilConnectedWait = Integer.parseInt(environment.getProperty("albedo.rpc.zookeeper.blockUntilConnectedWait",
            zookeeper.getBlockUntilConnectedWait()+""));
    logger.info("CuratorFramework namespace {}", namespace);
    CuratorFramework curator = builder
            .retryPolicy(retryPolicy)
            .canBeReadOnly(true)
            .namespace(namespace)
            .connectString(connectString)
            .defaultData(null)
            .build();
    curator.blockUntilConnected(blockUntilConnectedWait,
            TimeUnit.SECONDS);
    curator.start();
    return curator;
}

项目：pravega 文件：SegmentStoreAdapter.java

private ServiceBuilder attachDataLogFactory(ServiceBuilder builder) {
    if (this.config.getBookieCount() > 0) {
        // We were instructed to start at least one Bookie.
        this.zkClient = CuratorFrameworkFactory
                .builder()
                .connectString("localhost:" + this.config.getZkPort())
                .namespace("pravega")
                .retryPolicy(new ExponentialBackoffRetry(1000, 5))
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(5000)
                .build();
        this.zkClient.start();
        return builder.withDataLogFactory(setup -> {
            BookKeeperConfig bkConfig = setup.getConfig(BookKeeperConfig::builder);
            return new BookKeeperLogFactory(bkConfig, this.zkClient, setup.getCoreExecutor());
        });
    } else {
        // No Bookies -> InMemory Tier1.
        return builder.withDataLogFactory(setup -> new InMemoryDurableDataLogFactory(setup.getCoreExecutor()));
    }
}

项目：Distributed-Kit 文件：ZkReentrantLockTemplateTest.java

public static void main(String[] args){
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", retryPolicy);
    client.start();

    final ZkDistributedLockTemplate template=new ZkDistributedLockTemplate(client);//本类多线程安全,可通过spring注入
    template.execute("订单流水号", 5000, new Callback() {
        @Override
        public Object onGetLock() throws InterruptedException {
            //TODO 获得锁后要做的事
            return null;
        }

        @Override
        public Object onTimeout() throws InterruptedException {
            //TODO 获得锁超时后要做的事
            return null;
        }
    });
}

项目：aliyun-oss-hadoop-fs 文件：ZKRMStateStore.java

private void createConnection() throws Exception {
  // Curator connection
  CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
  builder = builder.connectString(zkHostPort)
      .connectionTimeoutMs(zkSessionTimeout)
      .retryPolicy(new RetryNTimes(numRetries, zkRetryInterval));

  // Set up authorization based on fencing scheme
  List<AuthInfo> authInfos = new ArrayList<>();
  for (ZKUtil.ZKAuthInfo zkAuth : zkAuths) {
    authInfos.add(new AuthInfo(zkAuth.getScheme(), zkAuth.getAuth()));
  }
  if (useDefaultFencingScheme) {
    byte[] defaultFencingAuth =
        (zkRootNodeUsername + ":" + zkRootNodePassword).getBytes(
            Charset.forName("UTF-8"));
    authInfos.add(new AuthInfo(zkRootNodeAuthScheme, defaultFencingAuth));
  }
  builder = builder.authorization(authInfos);

  // Connect to ZK
  curatorFramework = builder.build();
  curatorFramework.start();
}

项目：OSP-sample 文件：CacheTest.java

public void testNodeCache() throws Exception {
    CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181,localhost:3181,localhost:4181", new ExponentialBackoffRetry(1000,4));
    client.start();

    final NodeCache nodeCache = new NodeCache(client , "/nodecache/xiayiji");
    nodeCache.getListenable().addListener(new NodeCacheListener() {
        public void nodeChanged() throws Exception {
            if (nodeCache.getCurrentData() == null) {
                System.out.println("节点被删除了");
            } else {
                System.out.println(nodeCache.getCurrentData().getPath());
                System.out.println(nodeCache.getCurrentData().getStat());
                System.out.println(new String(nodeCache.getCurrentData().getData()));
                System.out.println("------------------------");
            }
        }
    });
    nodeCache.start(true);

    new BufferedReader(new InputStreamReader(System.in)).readLine();
}

项目：aliyun-oss-hadoop-fs 文件：RegistrySecurity.java

/**
 * Apply the security environment to this curator instance. This
 * may include setting up the ZK system properties for SASL
 * @param builder curator builder
 */
public void applySecurityEnvironment(CuratorFrameworkFactory.Builder builder) {

  if (isSecureRegistry()) {
    switch (access) {
      case anon:
        clearZKSaslClientProperties();
        break;

      case digest:
        // no SASL
        clearZKSaslClientProperties();
        builder.authorization(SCHEME_DIGEST, digestAuthData);
        break;

      case sasl:
        // bind to the current identity and context within the JAAS file
        setZKSaslClientProperties(jaasClientIdentity, jaasClientContext);
    }
  }
}

项目：newtranx-utils 文件：ZkPropertiesFactoryBean.java

@Override
protected Properties createProperties() throws IOException {
    Properties prop = super.createProperties();
    if (zkLocation == null || "".equals(zkLocation))
        zkLocation = environment.getProperty(this.zkLocationEnvName);
    if (zkLocation == null) {
        logger.warn("ZK connection string env var does not exist, skip zk based configuration.");
        return prop;
    }
    try {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(zkLocation, retryPolicy);
        client.start();
        int count = CuratorUtils.loadToProperties(client, prop, "/");
        logger.info(String.format("Loaded %d properties from zookeeper.", count));
        client.close();
    } catch (Exception ex) {
        logger.warn("Failed to load configuration from ZK.", ex);
    }
    return prop;
}

项目：pravega 文件：InProcPravegaCluster.java

private void cleanUpZK() {
    String[] pathsTobeCleaned = {"/pravega", "/hostIndex", "/store", "/taskIndex"};

    RetryPolicy rp = new ExponentialBackoffRetry(1000, 3);
    CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
            .connectString(zkUrl)
            .connectionTimeoutMs(5000)
            .sessionTimeoutMs(5000)
            .retryPolicy(rp);
    @Cleanup
    CuratorFramework zclient = builder.build();
    zclient.start();
    for ( String path : pathsTobeCleaned ) {
        try {
            zclient.delete().guaranteed().deletingChildrenIfNeeded()
                    .forPath(path);
        } catch (Exception e) {
            log.warn("Not able to delete path {} . Exception {}", path, e.getMessage());
        }
    }
    zclient.close();
}

项目：elastic-jobx 文件：ZookeeperRegistryCenterModifyTest.java

@Test
public void assertPersistEphemeralSequential() throws Exception {
    zkRegCenter.persistEphemeralSequential("/sequential/test_sequential");
    zkRegCenter.persistEphemeralSequential("/sequential/test_sequential");
    CuratorFramework client = CuratorFrameworkFactory.newClient(ZK_CONNECTION_STRING, new RetryOneTime(2000));
    client.start();
    client.blockUntilConnected();
    List<String> actual = client.getChildren().forPath("/" + ZookeeperRegistryCenterModifyTest.class.getName() + "/sequential");
    assertThat(actual.size(), is(2));
    for (String each : actual) {
        assertThat(each, startsWith("test_sequential"));
    }
    zkRegCenter.close();
    actual = client.getChildren().forPath("/" + ZookeeperRegistryCenterModifyTest.class.getName() + "/sequential");
    assertTrue(actual.isEmpty());
    zkRegCenter.init();
}

项目：Microservices-Deployment-Cookbook 文件：ZookeeperServiceDiscovery.java

private static ServiceProvider<Object> getGeolocationServiceProvider() throws Exception {
    if(geolocationServiceProvider == null) {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("192.168.99.100:2181", new RetryNTimes(5, 1000));
        curatorFramework.start();

        ServiceDiscovery<Object> serviceDiscovery = ServiceDiscoveryBuilder.builder(Object.class)
                .basePath("com.packt.microservices")
                .client(curatorFramework)
                .build();
        serviceDiscovery.start();

        geolocationServiceProvider = serviceDiscovery.serviceProviderBuilder()
                .serviceName("geolocation")
                .build();
        geolocationServiceProvider.start();
    }
    return geolocationServiceProvider;
}

项目：x-pipe 文件：DefaultZkConfig.java

@Override
public CuratorFramework create(String address) throws InterruptedException {

    Builder builder = CuratorFrameworkFactory.builder();
    builder.connectionTimeoutMs(getZkConnectionTimeoutMillis());
    builder.connectString(address);
    builder.maxCloseWaitMs(getZkCloseWaitMillis());
    builder.namespace(getZkNamespace());
    builder.retryPolicy(new RetryNTimes(getZkRetries(), getSleepMsBetweenRetries()));
    builder.sessionTimeoutMs(getZkSessionTimeoutMillis());
    builder.threadFactory(XpipeThreadFactory.create("Xpipe-ZK-" + address, true));

    logger.info("[create]{}, {}", Codec.DEFAULT.encode(this), address);
    CuratorFramework curatorFramework = builder.build();
    curatorFramework.start();
    curatorFramework.blockUntilConnected(waitForZkConnectedMillis(), TimeUnit.MILLISECONDS);

    return curatorFramework;
}

项目：elastic-config 文件：ElasticConfigInitialScript.java

public static void main(String[] args) throws Exception {

        @Cleanup
        CuratorFramework client = CuratorFrameworkFactory.newClient(ZK,
                new ExponentialBackoffRetry(100, 2));
        client.start();

        for (Entry<String, String> item : data.entrySet()) {
            Stat stat = client.checkExists().forPath(item.getKey());
            if (stat == null) {
                client.create().creatingParentsIfNeeded()
                        .forPath(item.getKey(), item.getValue().getBytes());
            }
        }

        client.setData().forPath("/github/projectname",
                sha1Digest("123456").getBytes());

    }

项目：pravega 文件：ControllerServiceTest.java

public ControllerServiceTest() throws Exception {
    zkServer = new TestingServerStarter().start();
    zkServer.start();
    zkClient = CuratorFrameworkFactory.newClient(zkServer.getConnectString(),
            new ExponentialBackoffRetry(200, 10, 5000));
    zkClient.start();

    final TaskMetadataStore taskMetadataStore = TaskStoreFactory.createZKStore(zkClient, executor);
    final HostControllerStore hostStore = HostStoreFactory.createInMemoryStore(HostMonitorConfigImpl.dummyConfig());

    SegmentHelper segmentHelper = SegmentHelperMock.getSegmentHelperMock();
    connectionFactory = new ConnectionFactoryImpl(false);
    streamMetadataTasks = new StreamMetadataTasks(streamStore, hostStore,
            taskMetadataStore, segmentHelper, executor, "host", connectionFactory);
    streamTransactionMetadataTasks = new StreamTransactionMetadataTasks(streamStore,
            hostStore, segmentHelper, executor, "host", connectionFactory);

    consumer = new ControllerService(streamStore, hostStore, streamMetadataTasks, streamTransactionMetadataTasks,
            new SegmentHelper(), executor, null);
}

项目：bd-codes 文件：CuratorTestingServer.java

/**
 * 测试入口
 */
public static void main(String[] args) throws Exception {
    String path = "/zookeeper";

    TestingServer server = new TestingServer(2181, new File(
            "/download/zk/data"));

    CuratorFramework client = CuratorFrameworkFactory.builder()
            .connectString(server.getConnectString())
            .sessionTimeoutMs(50000).connectionTimeoutMs(100000)
            .retryPolicy(new ExponentialBackoffRetry(1000, 5)).build();
    client.start();

    System.out.println(client.getChildren().forPath(path));

    server.close();
}

项目：vertx-service-discovery 文件：ZookeeperBackendService.java

@Override
public void init(Vertx vertx, JsonObject configuration) {
  this.vertx = vertx;
  String connection = Objects.requireNonNull(configuration.getString("connection"));
  int maxRetries = configuration.getInteger("maxRetries", 3);
  int baseGraceBetweenRetries = configuration
      .getInteger("baseSleepTimeBetweenRetries", 1000);
  canBeReadOnly = configuration.getBoolean("canBeReadOnly", false);
  connectionTimeoutMs = configuration.getInteger("connectionTimeoutMs", 1000);
  basePath = configuration.getString("basePath", "/services");
  ephemeral = configuration.getBoolean("ephemeral", false);
  guaranteed = configuration.getBoolean("guaranteed", false);

  client = CuratorFrameworkFactory.builder()
      .canBeReadOnly(canBeReadOnly)
      .connectString(connection)
      .connectionTimeoutMs(connectionTimeoutMs)
      .retryPolicy(new ExponentialBackoffRetry(baseGraceBetweenRetries, maxRetries))
      .build();
  client.getConnectionStateListenable().addListener(this);
  client.start();
}

项目：maelstrom 文件：OffsetManager.java

public static CuratorFramework createCurator(final String zookeeper) {
    CuratorFramework curator = CuratorFrameworkFactory.builder().
            connectString(zookeeper).
            sessionTimeoutMs(120000).
            connectionTimeoutMs(120000).
            retryPolicy(new ExponentialBackoffRetry(1000, 29))
            .threadFactory(new ThreadFactory() {
                @Override
                public Thread newThread(@SuppressWarnings("NullableProblems") Runnable r) {
                    Thread t = new Thread(r);
                    t.setDaemon(true);
                    return t;
                }
            }).build();
    curator.start();

    return curator;
}