import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer; //导入依赖的package包/类
private void init() {
    zkClient = new ZkClient(url, sessionTimeout, connectionTimeout,
            new BytesPushThroughSerializer());
    try {
        zkClient.createPersistent(HOME.concat(root).concat("/").
                concat(parent), true);
    } catch (ZkNodeExistsException e) {
        e.printStackTrace();
    }
    cleanExecutors = Executors.newCachedThreadPool();

}
 

import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer; //导入依赖的package包/类
/**
 * Creates a connection to a Pinot cluster, given its Zookeeper URL
 *
 * @param zkUrl The URL to the Zookeeper cluster
 * @return A connection that connects to the brokers in the given Helix cluster
 */
public static Connection fromZookeeper(String zkUrl) {
  try {
    ZkClient client = new ZkClient(zkUrl);
    client.setZkSerializer(new BytesPushThroughSerializer());
    byte[] brokerResourceNodeData = client.readData("/EXTERNALVIEW/brokerResource", true);
    JSONObject jsonObject = new JSONObject(new String(brokerResourceNodeData));
    JSONObject brokerResourceNode = jsonObject.getJSONObject("mapFields");

    List<String> brokerUrls = new ArrayList<String>();

    Iterator<String> resourceNames = brokerResourceNode.keys();
    while (resourceNames.hasNext()) {
      String resourceName = resourceNames.next();
      JSONObject resource = brokerResourceNode.getJSONObject(resourceName);

      Iterator<String> brokerNames = resource.keys();
      while (brokerNames.hasNext()) {
        String brokerName = brokerNames.next();
        if (brokerName.startsWith("Broker_") && "ONLINE".equals(resource.getString(brokerName))) {
          // Turn Broker_12.34.56.78_1234 into 12.34.56.78:1234
          brokerUrls.add(brokerName.replace("Broker_", "").replace("_", ":"));
        }
      }
    }

    client.close();

    return new Connection(brokerUrls, _transportFactory.buildTransport());
  } catch (Exception e) {
    throw new PinotClientException(e);
  }
}
 

import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer; //导入依赖的package包/类
public DynamicBrokerSelector(String zkServers) {
  ZkClient zkClient = new ZkClient(zkServers);
  zkClient.setZkSerializer(new BytesPushThroughSerializer());
  zkClient.waitUntilConnected(60, TimeUnit.SECONDS);
  zkClient.subscribeDataChanges(ExternalViewReader.BROKER_EXTERNAL_VIEW_PATH, this);
  evReader = new ExternalViewReader(zkClient);
  refresh();
}
 

import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer; //导入依赖的package包/类
/**
 * Create new ZkClient of servers
 *
 * @param servers ZooKeeper client hosts
 */
public ZKClient(String servers) {
  client = new ZkClient(servers, SESSION_TIMEOUT, CONNECTION_TIMEOUT, new BytesPushThroughSerializer());
}
 

import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer; //导入依赖的package包/类
private ZkClient client() { return new ZkClient(connect, 30000, 30000, new BytesPushThroughSerializer()); }