本文整理了Java中org.apache.twill.zookeeper.ZKClientServices类的一些代码示例，展示了ZKClientServices类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ZKClientServices类的具体详情如下：
包路径：org.apache.twill.zookeeper.ZKClientServices
类名称：ZKClientServices

ZKClientServices介绍

[英]Provides static factory method to create ZKClientService with modified behaviors.
[中]提供静态工厂方法来创建具有修改行为的ZKClientService。

代码示例

代码示例来源：origin: apache/phoenix

TxConstants.Service.DEFAULT_DATA_TX_CLIENT_DISCOVERY_TIMEOUT_SEC);
ZKClientService zkClientService  = ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(

代码示例来源：origin: apache/phoenix

@Override
public PhoenixTransactionService getTransactionService(Configuration config, ConnectionInfo connInfo, int port) {
  config.setInt(TxConstants.Service.CFG_DATA_TX_BIND_PORT, port);
  int retryTimeOut = config.getInt(TxConstants.Service.CFG_DATA_TX_CLIENT_DISCOVERY_TIMEOUT_SEC, 
      TxConstants.Service.DEFAULT_DATA_TX_CLIENT_DISCOVERY_TIMEOUT_SEC);
  ZKClientService zkClient = ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(connInfo.getZookeeperConnectionString())
      .setSessionTimeout(config.getInt(HConstants.ZK_SESSION_TIMEOUT,
          HConstants.DEFAULT_ZK_SESSION_TIMEOUT))
      .build(),
     RetryStrategies.exponentialDelay(500, retryTimeOut, TimeUnit.MILLISECONDS)
    )
   )
  );
  DiscoveryService discovery = new ZKDiscoveryService(zkClient);
  TransactionManager txManager = new TransactionManager(config, new HDFSTransactionStateStorage(config, 
      new SnapshotCodecProvider(config), new TxMetricsCollector()), new TxMetricsCollector());
  TransactionService txService = new TransactionService(config, zkClient, discovery, Providers.of(txManager));
  TephraTransactionService service = new TephraTransactionService(zkClient, txService);
  service.start();
  return service;
}

代码示例来源：origin: org.apache.twill/twill-zookeeper

/**
 * Creates a {@link ZKClientService} from the given {@link ZKClient} if the given {@link ZKClient} is an instance of
 * {@link ZKClientService} or is a {@link ForwardingZKClient} that eventually trace back to a delegate of type
 * {@link ZKClientService}. If such a {@link ZKClientService} instance is found, this method returns
 * an instance by invoking {@link #delegate(ZKClient, ZKClientService)} with the given {@link ZKClient} and
 * the {@link ZKClientService} found respectively.
 *
 * @param client The {@link ZKClient}.
 * @return A {@link ZKClientService}.
 * @throws IllegalArgumentException If no {@link ZKClientService} is found.
 */
public static ZKClientService delegate(ZKClient client) {
 ZKClient zkClient = client;
 while (!(zkClient instanceof ZKClientService) && zkClient instanceof ForwardingZKClient) {
  zkClient = ((ForwardingZKClient) zkClient).getDelegate();
 }
 if (zkClient instanceof ZKClientService) {
  return delegate(client, (ZKClientService) zkClient);
 }
 throw new IllegalArgumentException("No ZKClientService found from the delegation hierarchy");
}

代码示例来源：origin: apache/twill

/**
 * Creates a {@link ZKClientService} from the given {@link ZKClient} if the given {@link ZKClient} is an instance of
 * {@link ZKClientService} or is a {@link ForwardingZKClient} that eventually trace back to a delegate of type
 * {@link ZKClientService}. If such a {@link ZKClientService} instance is found, this method returns
 * an instance by invoking {@link #delegate(ZKClient, ZKClientService)} with the given {@link ZKClient} and
 * the {@link ZKClientService} found respectively.
 *
 * @param client The {@link ZKClient}.
 * @return A {@link ZKClientService}.
 * @throws IllegalArgumentException If no {@link ZKClientService} is found.
 */
public static ZKClientService delegate(ZKClient client) {
 ZKClient zkClient = client;
 while (!(zkClient instanceof ZKClientService) && zkClient instanceof ForwardingZKClient) {
  zkClient = ((ForwardingZKClient) zkClient).getDelegate();
 }
 if (zkClient instanceof ZKClientService) {
  return delegate(client, (ZKClientService) zkClient);
 }
 throw new IllegalArgumentException("No ZKClientService found from the delegation hierarchy");
}

代码示例来源：origin: co.cask.tephra/tephra-core

@Provides
 @Singleton
 private ZKClientService provideZKClientService(Configuration conf) {
  String zkStr = conf.get(TxConstants.Service.CFG_DATA_TX_ZOOKEEPER_QUORUM);
  if (zkStr == null) {
   // Default to HBase one.
   zkStr = conf.get(TxConstants.HBase.ZOOKEEPER_QUORUM);
  }

  int timeOut = conf.getInt(TxConstants.HBase.ZK_SESSION_TIMEOUT, TxConstants.HBase.DEFAULT_ZK_SESSION_TIMEOUT);
  ZKClientService zkClientService = new TephraZKClientService(zkStr, timeOut, null,
                                ArrayListMultimap.<String, byte[]>create());
  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(zkClientService, RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: caskdata/tephra

@Provides
 @Singleton
 private ZKClientService provideZKClientService(Configuration conf) {
  String zkStr = conf.get(TxConstants.Service.CFG_DATA_TX_ZOOKEEPER_QUORUM);
  if (zkStr == null) {
   // Default to HBase one.
   zkStr = conf.get(TxConstants.HBase.ZOOKEEPER_QUORUM);
  }

  int timeOut = conf.getInt(TxConstants.HBase.ZK_SESSION_TIMEOUT, TxConstants.HBase.DEFAULT_ZK_SESSION_TIMEOUT);
  ZKClientService zkClientService = new TephraZKClientService(zkStr, timeOut, null,
                                ArrayListMultimap.<String, byte[]>create());
  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(zkClientService, RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: org.apache.tephra/tephra-core

@Provides
 @Singleton
 private ZKClientService provideZKClientService(Configuration conf) {
  String zkStr = conf.get(TxConstants.Service.CFG_DATA_TX_ZOOKEEPER_QUORUM);
  if (zkStr == null) {
   // Default to HBase one.
   zkStr = conf.get(TxConstants.HBase.ZOOKEEPER_QUORUM);
  }

  int timeOut = conf.getInt(TxConstants.HBase.ZK_SESSION_TIMEOUT, TxConstants.HBase.DEFAULT_ZK_SESSION_TIMEOUT);
  ZKClientService zkClientService = new TephraZKClientService(zkStr, timeOut, null,
                                ArrayListMultimap.<String, byte[]>create());
  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(zkClientService, RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: co.cask.cdap/cdap-common

@Provides
 @Singleton
 private ZKClientService provideZKClientService(CConfiguration cConf) {
  String zookeeper = cConf.get(Constants.Zookeeper.QUORUM);
  Preconditions.checkNotNull(zookeeper, "Missing ZooKeeper configuration '%s'", Constants.Zookeeper.QUORUM);

  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(cConf.get(Constants.Zookeeper.QUORUM))
      .setSessionTimeout(cConf.getInt(Constants.Zookeeper.CFG_SESSION_TIMEOUT_MILLIS,
                      Constants.Zookeeper.DEFAULT_SESSION_TIMEOUT_MILLIS))
      .build(),
     RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: caskdata/cdap

@Provides
 @Singleton
 private ZKClientService provideZKClientService(CConfiguration cConf) {
  String zookeeper = cConf.get(Constants.Zookeeper.QUORUM);
  Preconditions.checkNotNull(zookeeper, "Missing ZooKeeper configuration '%s'", Constants.Zookeeper.QUORUM);

  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(cConf.get(Constants.Zookeeper.QUORUM))
      .setSessionTimeout(cConf.getInt(Constants.Zookeeper.CFG_SESSION_TIMEOUT_MILLIS,
                      Constants.Zookeeper.DEFAULT_SESSION_TIMEOUT_MILLIS))
      .build(),
     RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: co.cask.tigon/tigon-yarn

@Provides
 @Singleton
 private ZKClientService provideZKClientService(CConfiguration cConf) {
  String zookeeper = cConf.get(Constants.Zookeeper.QUORUM);
  Preconditions.checkNotNull(zookeeper, "Missing Zookeeper configuration '%s'", Constants.Zookeeper.QUORUM);

  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(cConf.get(Constants.Zookeeper.QUORUM))
      .setSessionTimeout(cConf.getInt(Constants.Zookeeper.CFG_SESSION_TIMEOUT_MILLIS,
                      Constants.Zookeeper.DEFAULT_SESSION_TIMEOUT_MILLIS))
      .build(),
     RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: co.cask.cdap/cdap-common

kafkaNamespace = "/" + kafkaNamespace;
  zkClientService = ZKClientServices.delegate(ZKClients.namespace(zkClientService, kafkaNamespace));
} else {
 zkClientService = ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(

代码示例来源：origin: caskdata/cdap

kafkaNamespace = "/" + kafkaNamespace;
  zkClientService = ZKClientServices.delegate(ZKClients.namespace(zkClientService, kafkaNamespace));
} else {
 zkClientService = ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(

代码示例来源：origin: org.apache.twill/twill-yarn

private ZKClientService getZKClientService(String zkConnect) {
 return ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(ZKClientService.Builder.of(zkConnect)
                 .setSessionTimeout(ZK_TIMEOUT)
                 .build(), RetryStrategies.exponentialDelay(100, 2000, TimeUnit.MILLISECONDS))));
}

代码示例来源：origin: apache/twill

private ZKClientService getZKClientService(String zkConnect) {
 return ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(ZKClientService.Builder.of(zkConnect)
                 .setSessionTimeout(ZK_TIMEOUT)
                 .build(), RetryStrategies.exponentialDelay(100, 2000, TimeUnit.MILLISECONDS))));
}

代码示例来源：origin: org.apache.twill/twill-core

@Override
public void start() {
 Preconditions.checkNotNull(zkConnectStr);
 eventConverter = new LogEventConverter(hostname, runnableName);
 scheduler = Executors.newSingleThreadScheduledExecutor(Threads.createDaemonThreadFactory(PUBLISH_THREAD_NAME));
 zkClientService = ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(ZKClientService.Builder.of(zkConnectStr).build(),
                RetryStrategies.fixDelay(1, TimeUnit.SECONDS))));
 kafkaClient = new ZKKafkaClientService(zkClientService);
 Futures.addCallback(Services.chainStart(zkClientService, kafkaClient),
           new FutureCallback<List<ListenableFuture<Service.State>>>() {
  @Override
  public void onSuccess(List<ListenableFuture<Service.State>> result) {
   for (ListenableFuture<Service.State> future : result) {
    Preconditions.checkState(Futures.getUnchecked(future) == Service.State.RUNNING,
                 "Service is not running.");
   }
   addInfo("Kafka client started: " + zkConnectStr);
   scheduler.scheduleWithFixedDelay(flushTask, 0, flushPeriod, TimeUnit.MILLISECONDS);
  }
  @Override
  public void onFailure(Throwable t) {
   // Fail to talk to kafka. Other than logging, what can be done?
   addError("Failed to start kafka appender.", t);
  }
 }, Threads.SAME_THREAD_EXECUTOR);
 super.start();
}

代码示例来源：origin: apache/twill

@Override
public void start() {
 Preconditions.checkNotNull(zkConnectStr);
 eventConverter = new LogEventConverter(hostname, runnableName);
 scheduler = Executors.newSingleThreadScheduledExecutor(Threads.createDaemonThreadFactory(PUBLISH_THREAD_NAME));
 zkClientService = ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(ZKClientService.Builder.of(zkConnectStr).build(),
                RetryStrategies.fixDelay(1, TimeUnit.SECONDS))));
 kafkaClient = new ZKKafkaClientService(zkClientService);
 Futures.addCallback(Services.chainStart(zkClientService, kafkaClient),
           new FutureCallback<List<ListenableFuture<Service.State>>>() {
  @Override
  public void onSuccess(List<ListenableFuture<Service.State>> result) {
   for (ListenableFuture<Service.State> future : result) {
    Preconditions.checkState(Futures.getUnchecked(future) == Service.State.RUNNING,
                 "Service is not running.");
   }
   addInfo("Kafka client started: " + zkConnectStr);
   scheduler.scheduleWithFixedDelay(flushTask, 0, flushPeriod, TimeUnit.MILLISECONDS);
  }
  @Override
  public void onFailure(Throwable t) {
   // Fail to talk to kafka. Other than logging, what can be done?
   addError("Failed to start kafka appender.", t);
  }
 }, Threads.SAME_THREAD_EXECUTOR);
 super.start();
}

代码示例来源：origin: com.aliyun.phoenix/ali-phoenix-core

TxConstants.Service.DEFAULT_DATA_TX_CLIENT_DISCOVERY_TIMEOUT_SEC);
ZKClientService zkClientService  = ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(

代码示例来源：origin: org.apache.phoenix/phoenix-core

ZKClientService zkClientService  = ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(

代码示例来源：origin: caskdata/coopr

private ZKClientService getZKService(String connectString) {
  return ZKClientServices.delegate(
   ZKClients.namespace(
    ZKClients.reWatchOnExpire(
     ZKClients.retryOnFailure(
      ZKClientService.Builder.of(connectString)
       .setSessionTimeout(conf.getInt(Constants.ZOOKEEPER_SESSION_TIMEOUT_MILLIS))
       .build(),
      RetryStrategies.fixDelay(2, TimeUnit.SECONDS)
     )
    ), conf.get(Constants.ZOOKEEPER_NAMESPACE)
   )
  );
 }
}

代码示例来源：origin: org.apache.twill/twill-yarn

/**
 * Creates a {@link ZKClientService}.
 */
protected final ZKClientService createZKClient() {
 TwillRuntimeSpecification twillRuntimeSpec = getTwillRuntimeSpecification();
 return ZKClientServices.delegate(
  ZKClients.namespace(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(twillRuntimeSpec.getZkConnectStr()).build(),
     RetryStrategies.fixDelay(1, TimeUnit.SECONDS)
    )
   ), "/" + twillRuntimeSpec.getTwillAppName()
  )
 );
}