import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
/**
 * 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);
    }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
/**
 * Create a node.
 * 
 * @param path
 * @param data
 * @param mode
 * @return create node's path
 * @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 create(final String path, Object data, final CreateMode mode) throws ZkInterruptedException,
                                                                           IllegalArgumentException, ZkException,
                                                                           RuntimeException {
    if (path == null) {
        throw new NullPointerException("path must not be null.");
    }
    final byte[] bytes = data == null ? null : serialize(data);

    return retryUntilConnected(new Callable<String>() {

        @Override
        public String call() throws Exception {
            return _connection.create(path, bytes, mode);
        }
    });
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
public boolean waitUntilExists(String path, TimeUnit timeUnit, long time) throws ZkInterruptedException {
    Date timeout = new Date(System.currentTimeMillis() + timeUnit.toMillis(time));
    LOG.debug("Waiting until znode '" + path + "' becomes available.");
    if (exists(path)) {
        return true;
    }
    acquireEventLock();
    try {
        while (!exists(path, true)) {
            boolean gotSignal = getEventLock().getZNodeEventCondition().awaitUntil(timeout);
            if (!gotSignal) {
                return false;
            }
        }
        return true;
    } catch (InterruptedException e) {
        throw new ZkInterruptedException(e);
    } finally {
        getEventLock().unlock();
    }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
public boolean waitForKeeperState(KeeperState keeperState, long time, TimeUnit timeUnit)
                                                                                        throws ZkInterruptedException {
    if (_zookeeperEventThread != null && Thread.currentThread() == _zookeeperEventThread) {
        throw new IllegalArgumentException("Must not be done in the zookeeper event thread.");
    }
    Date timeout = new Date(System.currentTimeMillis() + timeUnit.toMillis(time));

    LOG.debug("Waiting for keeper state " + keeperState);
    acquireEventLock();
    try {
        boolean stillWaiting = true;
        while (_currentState != keeperState) {
            if (!stillWaiting) {
                return false;
            }
            stillWaiting = getEventLock().getStateChangedCondition().awaitUntil(timeout);
        }
        LOG.debug("State is " + _currentState);
        return true;
    } catch (InterruptedException e) {
        throw new ZkInterruptedException(e);
    } finally {
        getEventLock().unlock();
    }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
/**
 * Close the client.
 * 
 * @throws ZkInterruptedException
 */
public void close() throws ZkInterruptedException {
    if (_connection == null) {
        return;
    }
    LOG.debug("Closing ZkClient...");
    getEventLock().lock();
    try {
        setShutdownTrigger(true);
        _eventThread.interrupt();
        _eventThread.join(2000);
        _connection.close();
        _connection = null;
    } catch (InterruptedException e) {
        throw new ZkInterruptedException(e);
    } finally {
        getEventLock().unlock();
    }
    LOG.debug("Closing ZkClient...done");
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
protected boolean isInterrupt(Throwable e) {
    if (!running) {
        return true;
    }

    if (e instanceof InterruptedException || e instanceof ZkInterruptedException) {
        return true;
    }

    if (ExceptionUtils.getRootCause(e) instanceof InterruptedException) {
        return true;
    }

    return false;

}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
public boolean waitUntilExists(String path, TimeUnit timeUnit, long time)
    throws ZkInterruptedException {
  Date timeout = new Date(System.currentTimeMillis() + timeUnit.toMillis(time));
  LOG.debug("Waiting until znode '" + path + "' becomes available.");
  if (exists(path)) {
    return true;
  }
  acquireEventLock();
  try {
    while (!exists(path, true)) {
      boolean gotSignal = getEventLock().getZNodeEventCondition().awaitUntil(timeout);
      if (!gotSignal) {
        return false;
      }
    }
    return true;
  } catch (InterruptedException e) {
    throw new ZkInterruptedException(e);
  } finally {
    getEventLock().unlock();
  }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
public boolean waitForKeeperState(KeeperState keeperState, long time, TimeUnit timeUnit)
    throws ZkInterruptedException {
  if (_zookeeperEventThread != null && Thread.currentThread() == _zookeeperEventThread) {
    throw new IllegalArgumentException("Must not be done in the zookeeper event thread.");
  }
  Date timeout = new Date(System.currentTimeMillis() + timeUnit.toMillis(time));

  LOG.debug("Waiting for keeper state " + keeperState);
  acquireEventLock();
  try {
    boolean stillWaiting = true;
    while (_currentState != keeperState) {
      if (!stillWaiting) {
        return false;
      }
      stillWaiting = getEventLock().getStateChangedCondition().awaitUntil(timeout);
    }
    LOG.debug("State is " + _currentState);
    return true;
  } catch (InterruptedException e) {
    throw new ZkInterruptedException(e);
  } finally {
    getEventLock().unlock();
  }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
/**
 * 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);
    }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
private void acquireEventLock() {
    try {
        getEventLock().lockInterruptibly();
    } catch (InterruptedException e) {
        throw new ZkInterruptedException(e);
    }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
/**
 * Connect to ZooKeeper.
 * 
 * @param maxMsToWaitUntilConnected
 * @param watcher
 * @throws ZkInterruptedException if the connection timed out due to thread interruption
 * @throws ZkTimeoutException if the connection timed out
 * @throws IllegalStateException if the connection timed out due to thread interruption
 */
public void connect(final long maxMsToWaitUntilConnected, Watcher watcher) throws ZkInterruptedException,
                                                                          ZkTimeoutException, IllegalStateException {
    boolean started = false;
    try {
        getEventLock().lockInterruptibly();
        setShutdownTrigger(false);
        _eventThread = new ZkEventThread(_connection.getServers());
        _eventThread.start();
        _connection.connect(watcher);

        LOG.debug("Awaiting connection to Zookeeper server");
        if (!waitUntilConnected(maxMsToWaitUntilConnected, TimeUnit.MILLISECONDS)) {
            throw new ZkTimeoutException("Unable to connect to zookeeper server within timeout: "
                                         + maxMsToWaitUntilConnected);
        }
        started = true;
    } catch (InterruptedException e) {
        States state = _connection.getZookeeperState();
        throw new IllegalStateException("Not connected with zookeeper server yet. Current state is " + state);
    } finally {
        getEventLock().unlock();

        // we should close the zookeeper instance, otherwise it would keep
        // on trying to connect
        if (!started) {
            close();
        }
    }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
private void reconnect() {
    getEventLock().lock();
    try {
        _connection.close();
        _connection.connect(this);
    } catch (InterruptedException e) {
        throw new ZkInterruptedException(e);
    } finally {
        getEventLock().unlock();
    }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
public void connect() throws ZkInterruptedException {
  boolean isConnected = zkClient.waitUntilConnected(connectionTimeoutMs, TimeUnit.MILLISECONDS);
  if (!isConnected) {
    if (metrics != null) {
      metrics.zkConnectionError.inc();
    }
    throw new RuntimeException("Unable to connect to Zookeeper within connectionTimeout " + connectionTimeoutMs + "ms. Shutting down!");
  }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
public void close() {
  try {
    if (zkUtils != null)
      zkUtils.close();
  } catch (ZkInterruptedException ex) {
    // Swallowing due to occurrence in the last stage of lifecycle(Not actionable).
    LOG.error("Exception in close(): ", ex);
  }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
private void acquireEventLock() {
  try {
    getEventLock().lockInterruptibly();
  } catch (InterruptedException e) {
    throw new ZkInterruptedException(e);
  }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
/**
 * Connect to ZooKeeper.
 *
 * @param maxMsToWaitUntilConnected
 * @param watcher
 * @throws ZkInterruptedException
 *             if the connection timed out due to thread interruption
 * @throws ZkTimeoutException
 *             if the connection timed out
 * @throws IllegalStateException
 *             if the connection timed out due to thread interruption
 */
public void connect(final long maxMsToWaitUntilConnected, Watcher watcher)
    throws ZkInterruptedException, ZkTimeoutException, IllegalStateException {
  boolean started = false;
  acquireEventLock();
  try {
    setShutdownTrigger(false);
    _eventThread = new ZkEventThread(_connection.getServers());
    _eventThread.start();
    _connection.connect(watcher);

    LOG.debug("Awaiting connection to Zookeeper server");
    if (!waitUntilConnected(maxMsToWaitUntilConnected, TimeUnit.MILLISECONDS)) {
      throw new ZkTimeoutException(
          "Unable to connect to zookeeper server within timeout: " + maxMsToWaitUntilConnected);
    }
    started = true;
  } finally {
    getEventLock().unlock();

    // we should close the zookeeper instance, otherwise it would keep
    // on trying to connect
    if (!started) {
      close();
    }
  }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
private void reconnect() {
  getEventLock().lock();
  try {
    _connection.close();
    _connection.connect(this);
  } catch (InterruptedException e) {
    throw new ZkInterruptedException(e);
  } finally {
    getEventLock().unlock();
  }
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
public void waitUntilConnected() throws ZkInterruptedException {
    waitUntilConnected(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
@Override
public void createPersistent(String path, boolean createParents) throws ZkInterruptedException, IllegalArgumentException, ZkException, RuntimeException {
    zkClient.createPersistent(path, createParents);
}
 

import org.I0Itec.zkclient.exception.ZkInterruptedException; //导入依赖的package包/类
@Override
public void createEphemeral(String path, Object data) throws ZkInterruptedException, IllegalArgumentException, ZkException, RuntimeException {
    zkClient.createEphemeral(path, data);
}