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

ZkClientx介绍

[英]使用自定义的ZooKeeperx for zk connection
[中]使用自定义的zk连接用ZooKeeperx

代码示例

代码示例来源：origin: alibaba/canal

@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
  String path = ZookeeperPathUtils.getParsePath(destination);
  byte[] data = JsonUtils.marshalToByte(logPosition);
  try {
    zkClientx.writeData(path, data);
  } catch (ZkNoNodeException e) {
    zkClientx.createPersistent(path, data, true);
  }
}

代码示例来源：origin: alibaba/canal

private void releaseCid(String path) {
  // logger.info("## release the canalId = {}", cid);
  // 初始化系统目录
  if (zkclientx != null) {
    zkclientx.delete(path);
  }
}

代码示例来源：origin: alibaba/canal

public ClusterNodeAccessStrategy(String destination, ZkClientx zkClient){
  this.destination = destination;
  this.zkClient = zkClient;
  childListener = new IZkChildListener() {
    public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
      initClusters(currentChilds);
    }
  };
  dataListener = new IZkDataListener() {
    public void handleDataDeleted(String dataPath) throws Exception {
      runningAddress = null;
    }
    public void handleDataChange(String dataPath, Object data) throws Exception {
      initRunning(data);
    }
  };
  String clusterPath = ZookeeperPathUtils.getDestinationClusterRoot(destination);
  this.zkClient.subscribeChildChanges(clusterPath, childListener);
  initClusters(this.zkClient.getChildren(clusterPath));
  String runningPath = ZookeeperPathUtils.getDestinationServerRunning(destination);
  this.zkClient.subscribeDataChanges(runningPath, dataListener);
  initRunning(this.zkClient.readData(runningPath, true));
}

代码示例来源：origin: alibaba/canal

private void initCid(String path) {
  // logger.info("## init the canalId = {}", cid);
  // 初始化系统目录
  if (zkclientx != null) {
    try {
      zkclientx.createEphemeral(path);
    } catch (ZkNoNodeException e) {
      // 如果父目录不存在，则创建
      String parentDir = path.substring(0, path.lastIndexOf('/'));
      zkclientx.createPersistent(parentDir, true);
      zkclientx.createEphemeral(path);
    } catch (ZkNodeExistsException e) {
      // ignore
      // 因为第一次启动时创建了cid,但在stop/start的时可能会关闭和新建,允许出现NodeExists问题s
    }
  }
}

代码示例来源：origin: alibaba/canal

/**
 * Create a persistent Sequential node.
 *
 * @param path
 * @param createParents if true all parent dirs are created as well and no
 * {@link ZkNodeExistsException} is thrown in case the path already exists
 * @throws ZkInterruptedException if operation was interrupted, or a
 * required reconnection got interrupted
 * @throws IllegalArgumentException if called from anything except the
 * ZooKeeper event thread
 * @throws ZkException if any ZooKeeper exception occurred
 * @throws RuntimeException if any other exception occurs
 */
public String createPersistentSequential(String path, boolean createParents) throws ZkInterruptedException,
                                      IllegalArgumentException, ZkException,
                                      RuntimeException {
  try {
    return create(path, null, CreateMode.PERSISTENT_SEQUENTIAL);
  } catch (ZkNoNodeException e) {
    if (!createParents) {
      throw e;
    }
    String parentDir = path.substring(0, path.lastIndexOf('/'));
    createPersistent(parentDir, createParents);
    return createPersistentSequential(path, createParents);
  }
}

代码示例来源：origin: alibaba/canal

RuntimeException {
try {
  create(path, data, CreateMode.PERSISTENT);
} catch (ZkNodeExistsException e) {
  if (!createParents) {
  createPersistent(parentDir, createParents);
  createPersistent(path, data, createParents);

代码示例来源：origin: alibaba/canal

private void initRunning() {
  if (!isStart()) {
    return;
  }
  String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
  // 序列化
  byte[] bytes = JsonUtils.marshalToByte(serverData);
  try {
    mutex.set(false);
    zkClient.create(path, bytes, CreateMode.EPHEMERAL);
    activeData = serverData;
    processActiveEnter();// 触发一下事件
    mutex.set(true);
  } catch (ZkNodeExistsException e) {
    bytes = zkClient.readData(path, true);
    if (bytes == null) {// 如果不存在节点，立即尝试一次
      initRunning();
    } else {
      activeData = JsonUtils.unmarshalFromByte(bytes, ServerRunningData.class);
    }
  } catch (ZkNoNodeException e) {
    zkClient.createPersistent(ZookeeperPathUtils.getDestinationPath(destination), true); // 尝试创建父节点
    initRunning();
  }
}

代码示例来源：origin: alibaba/canal

zkclientx = ZkClientx.getZkClient(zkServers);
zkclientx.createPersistent(ZookeeperPathUtils.DESTINATION_ROOT_NODE, true);
zkclientx.createPersistent(ZookeeperPathUtils.CANAL_CLUSTER_ROOT_NODE, true);

代码示例来源：origin: alibaba/canal

@Override
public LogPosition getLatestIndexBy(String destination) {
  String path = ZookeeperPathUtils.getParsePath(destination);
  byte[] data = zkClientx.readData(path, true);
  if (data == null || data.length == 0) {
    return null;
  }
  return JsonUtils.unmarshalFromByte(data, LogPosition.class);
}

代码示例来源：origin: alibaba/canal

public void clearAllBatchs(ClientIdentity clientIdentity) throws CanalMetaManagerException {
  String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(),
    clientIdentity.getClientId());
  List<String> batchChilds = zkClientx.getChildren(path);
  for (String batchChild : batchChilds) {
    String batchPath = path + ZookeeperPathUtils.ZOOKEEPER_SEPARATOR + batchChild;
    zkClientx.delete(batchPath);
  }
}

代码示例来源：origin: alibaba/canal

List<String> childs = null;
try {
  childs = zkClientx.getChildren(path);
} catch (ZkNoNodeException e) {
for (Short clientId : clientIds) {
  path = ZookeeperPathUtils.getFilterPath(destination, clientId);
  byte[] bytes = zkClientx.readData(path, true);
  String filter = null;
  if (bytes != null) {

代码示例来源：origin: alibaba/canal

private synchronized ZkClientx getZkclientx() {
  // 做一下排序，保证相同的机器只使用同一个链接
  List<String> zkClusters = new ArrayList<String>(parameters.getZkClusters());
  Collections.sort(zkClusters);
  return ZkClientx.getZkClient(StringUtils.join(zkClusters, ";"));
}

代码示例来源：origin: alibaba/canal

public void addBatch(ClientIdentity clientIdentity, PositionRange positionRange,
           Long batchId) throws CanalMetaManagerException {
  String path = ZookeeperPathUtils
    .getBatchMarkWithIdPath(clientIdentity.getDestination(), clientIdentity.getClientId(), batchId);
  byte[] data = JsonUtils.marshalToByte(positionRange, SerializerFeature.WriteClassName);
  zkClientx.createPersistent(path, data, true);
}

代码示例来源：origin: alibaba/canal

private void processActiveEnter() {
  if (listener != null) {
    // 触发回调，建立与server的socket链接
    InetSocketAddress connectAddress = listener.processActiveEnter();
    String address = connectAddress.getAddress().getHostAddress() + ":" + connectAddress.getPort();
    this.clientData.setAddress(address);
    String path = ZookeeperPathUtils.getDestinationClientRunning(this.destination,
      this.clientData.getClientId());
    // 序列化
    byte[] bytes = JsonUtils.marshalToByte(clientData);
    zkClient.writeData(path, bytes);
  }
}

代码示例来源：origin: alibaba/canal

public void stop() {
  super.stop();
  String path = ZookeeperPathUtils.getDestinationClientRunning(this.destination, clientData.getClientId());
  zkClient.unsubscribeDataChanges(path, dataListener);
  releaseRunning(); // 尝试一下release
  // Fix issue #697
  if (delayExector != null) {
    delayExector.shutdown();
  }
}

代码示例来源：origin: alibaba/canal

public void start() {
  super.start();
  String path = ZookeeperPathUtils.getDestinationClientRunning(this.destination, clientData.getClientId());
  zkClient.subscribeDataChanges(path, dataListener);
  initRunning();
}

代码示例来源：origin: alibaba/canal

public Map<Long, PositionRange> listAllBatchs(ClientIdentity clientIdentity) {
  String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(),
    clientIdentity.getClientId());
  List<String> nodes = null;
  try {
    nodes = zkClientx.getChildren(path);
  } catch (ZkNoNodeException e) {
    // ignore
  }
  if (CollectionUtils.isEmpty(nodes)) {
    return Maps.newHashMap();
  }
  // 找到最大的Id
  ArrayList<Long> batchIds = new ArrayList<Long>(nodes.size());
  for (String batchIdString : nodes) {
    batchIds.add(Long.valueOf(batchIdString));
  }
  Collections.sort(batchIds); // 从小到大排序
  Map<Long, PositionRange> positionRanges = Maps.newLinkedHashMap();
  for (Long batchId : batchIds) {
    PositionRange result = getBatch(clientIdentity, batchId);
    if (result == null) {// 出现为null，说明zk节点有变化，重新获取
      return listAllBatchs(clientIdentity);
    } else {
      positionRanges.put(batchId, result);
    }
  }
  return positionRanges;
}

代码示例来源：origin: alibaba/canal

public Long addBatch(ClientIdentity clientIdentity, PositionRange positionRange) throws CanalMetaManagerException {
  String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(),
    clientIdentity.getClientId());
  byte[] data = JsonUtils.marshalToByte(positionRange, SerializerFeature.WriteClassName);
  String batchPath = zkClientx
    .createPersistentSequential(path + ZookeeperPathUtils.ZOOKEEPER_SEPARATOR, data, true);
  String batchIdString = StringUtils.substringAfterLast(batchPath, ZookeeperPathUtils.ZOOKEEPER_SEPARATOR);
  return ZookeeperPathUtils.getBatchMarkId(batchIdString);
}

代码示例来源：origin: alibaba/canal

try {
  mutex.set(false);
  zkClient.create(path, bytes, CreateMode.EPHEMERAL);
  bytes = zkClient.readData(path, true);
  if (bytes == null) {// 如果不存在节点，立即尝试一次
    initRunning();
  zkClient.createPersistent(ZookeeperPathUtils.getClientIdNodePath(this.destination, clientData.getClientId()),
    true); // 尝试创建父节点
  initRunning();

代码示例来源：origin: alibaba/canal

/**
 * Create a persistent Sequential node.
 *
 * @param path
 * @param data
 * @param createParents if true all parent dirs are created as well and no
 * {@link ZkNodeExistsException} is thrown in case the path already exists
 * @throws ZkInterruptedException if operation was interrupted, or a
 * required reconnection got interrupted
 * @throws IllegalArgumentException if called from anything except the
 * ZooKeeper event thread
 * @throws ZkException if any ZooKeeper exception occurred
 * @throws RuntimeException if any other exception occurs
 */
public String createPersistentSequential(String path, Object data, boolean createParents)
                                             throws ZkInterruptedException,
                                             IllegalArgumentException,
                                             ZkException,
                                             RuntimeException {
  try {
    return create(path, data, CreateMode.PERSISTENT_SEQUENTIAL);
  } catch (ZkNoNodeException e) {
    if (!createParents) {
      throw e;
    }
    String parentDir = path.substring(0, path.lastIndexOf('/'));
    createPersistent(parentDir, createParents);
    return createPersistentSequential(path, data, createParents);
  }
}