org.apache.pulsar.zookeeper.ZooKeeperChildrenCache类的使用及代码示例

x33g5p2x  于2022-02-05 转载在 其他  
字(11.0k)|赞(0)|评价(0)|浏览(113)

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

ZooKeeperChildrenCache介绍

[英]This class keeps a cache of the children for several z-nodes and updates it whenever it's changed on zookeeper.
[中]这个类为几个z节点保留子节点的缓存，并在zookeeper上进行更改时进行更新。

代码示例

代码示例来源：origin: org.apache.pulsar/pulsar-proxy

/**
 * Initialize ZooKeeper session and creates broker cache list
 *
 * @param zookeeperServers
 * @throws Exception
 */
public ZookeeperCacheLoader(ZooKeeperClientFactory factory, String zookeeperServers, int zookeeperSessionTimeoutMs) throws Exception {
  this.zkClient = factory.create(zookeeperServers, SessionType.AllowReadOnly, zookeeperSessionTimeoutMs).get();
  this.localZkCache = new LocalZooKeeperCache(zkClient, this.orderedExecutor);
  this.brokerInfo = new ZooKeeperDataCache<LoadManagerReport>(localZkCache) {
    @Override
    public LoadManagerReport deserialize(String key, byte[] content) throws Exception {
      return ObjectMapperFactory.getThreadLocal().readValue(content, LoadManagerReport.class);
    }
  };
  this.availableBrokersCache = new ZooKeeperChildrenCache(getLocalZkCache(), LOADBALANCE_BROKERS_ROOT);
  this.availableBrokersCache.registerListener((path, brokerNodes, stat) -> {
    try {
      updateBrokerList(brokerNodes);
    } catch (Exception e) {
      log.warn("Error updating broker info after broker list changed.", e);
    }
  });
  // Do initial fetch of brokers list
  try {
    updateBrokerList(availableBrokersCache.get());
  } catch (NoNodeException nne) { // can happen if no broker started yet
    updateBrokerList(Collections.emptySet());
  }
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

@Override
public void stop() throws PulsarServerException {
  loadReportCacheZk.close();
  loadReportCacheZk.clear();
  availableActiveBrokers.close();
  scheduler.shutdown();
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

@Override
public Set<String> getAvailableBrokers() throws Exception {
  return this.availableActiveBrokers.get();
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

availableActiveBrokers = new ZooKeeperChildrenCache(pulsar.getLocalZkCache(), LOADBALANCE_BROKERS_ROOT);
availableActiveBrokers.registerListener(new ZooKeeperCacheListener<Set<String>>() {
  @Override
  public void onUpdate(String path, Set<String> data, Stat stat) {

代码示例来源：origin: org.apache.pulsar/pulsar-broker-common

this.clustersListCache = new ZooKeeperChildrenCache(cache, CLUSTERS_ROOT);
  this.failureDomainListCache = new ZooKeeperChildrenCache(cache, CLUSTER_FAILURE_DOMAIN_ROOT);

代码示例来源：origin: org.apache.pulsar/pulsar-broker

@DELETE
@Path("/{cluster}/failureDomains/{domainName}")
@ApiOperation(value = "Delete cluster's failure omain")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission or plicy is read only"),
    @ApiResponse(code = 412, message = "Cluster doesn't exist") })
public void deleteFailureDomain(@PathParam("cluster") String cluster, @PathParam("domainName") String domainName)
    throws Exception {
  validateSuperUserAccess();
  validateClusterExists(cluster);
  try {
    final String domainPath = joinPath(pulsar().getConfigurationCache().CLUSTER_FAILURE_DOMAIN_ROOT, domainName);
    globalZk().delete(domainPath, -1);
    // clear domain cache
    failureDomainCache().invalidate(domainPath);
    failureDomainListCache().clear();
  } catch (KeeperException.NoNodeException nne) {
    log.warn("[{}] Domain {} does not exist in {}", clientAppId(), domainName, cluster);
    throw new RestException(Status.NOT_FOUND,
        "Domain-name " + domainName + " or cluster " + cluster + " does not exist");
  } catch (Exception e) {
    log.error("[{}] Failed to delete domain {} in cluster {}", clientAppId(), domainName, cluster, e);
    throw new RestException(e);
  }
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

@Override
public Set<String> getAvailableBrokers() {
  try {
    return availableActiveBrokers.get();
  } catch (Exception e) {
    log.warn("Error when trying to get active brokers", e);
    return loadData.getBrokerData().keySet();
  }
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

availableActiveBrokers = new ZooKeeperChildrenCache(pulsar.getLocalZkCache(),
    LoadManager.LOADBALANCE_BROKERS_ROOT);
availableActiveBrokers.registerListener(new ZooKeeperCacheListener<Set<String>>() {
  @Override
  public void onUpdate(String path, Set<String> data, Stat stat) {
    .registerListener((path, data, stat) -> scheduler.execute(() -> refreshBrokerToFailureDomainMap()));
pulsar.getConfigurationCache().failureDomainCache()
    .registerListener((path, data, stat) -> scheduler.execute(() -> refreshBrokerToFailureDomainMap()));

代码示例来源：origin: org.apache.pulsar/pulsar-broker

this.managedLedgerListCache = new ZooKeeperChildrenCache(cache, MANAGED_LEDGER_ROOT);
this.resourceQuotaCache = new ResourceQuotaCache(cache);
this.resourceQuotaCache.initZK();

代码示例来源：origin: org.apache.pulsar/pulsar-broker

if (this.createZnodeIfNotExist(domainPath, Optional.ofNullable(domain))) {
  this.failureDomainListCache().clear();
} else {
  globalZk().setData(domainPath, jsonMapper().writeValueAsBytes(domain), -1);

代码示例来源：origin: org.apache.pulsar/pulsar-discovery-service

this.availableBrokersCache = new ZooKeeperChildrenCache(getLocalZkCache(), LOADBALANCE_BROKERS_ROOT);
this.availableBrokersCache.registerListener((path, brokerNodes, stat) -> {
  try {
    updateBrokerList(brokerNodes);
updateBrokerList(availableBrokersCache.get());

代码示例来源：origin: org.apache.pulsar/pulsar-broker

@GET
@ApiOperation(value = "Get the list of all the Pulsar clusters.", response = String.class, responseContainer = "Set")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
public Set<String> getClusters() throws Exception {
  try {
    Set<String> clusters = clustersListCache().get();
    // Remove "global" cluster from returned list
    clusters.remove(Constants.GLOBAL_CLUSTER);
    return clusters;
  } catch (Exception e) {
    log.error("[{}] Failed to get clusters list", clientAppId(), e);
    throw new RestException(e);
  }
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

private void deleteFailureDomain(String clusterPath) {
  try {
    String failureDomain = joinPath(clusterPath, ConfigurationCacheService.FAILURE_DOMAIN);
    if (globalZk().exists(failureDomain, false) == null) {
      return;
    }
    for (String domain : globalZk().getChildren(failureDomain, false)) {
      String domainPath = joinPath(failureDomain, domain);
      globalZk().delete(domainPath, -1);
    }
    globalZk().delete(failureDomain, -1);
    failureDomainCache().clear();
    failureDomainListCache().clear();
  } catch (Exception e) {
    log.warn("Failed to delete failure-domain under cluster {}", clusterPath);
    throw new RestException(e);
  }
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

/**
 * As any broker, stop the load manager.
 *
 * @throws PulsarServerException
 *             If an unexpected error occurred when attempting to stop the load manager.
 */
@Override
public void stop() throws PulsarServerException {
  if (availableActiveBrokers != null) {
    availableActiveBrokers.close();
  }
  if (brokerDataCache != null) {
    brokerDataCache.close();
    brokerDataCache.clear();
  }
  scheduler.shutdown();
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

public List<String> getListOfPersistentTopics(NamespaceName namespaceName) throws Exception {
  List<String> topics = Lists.newArrayList();
  // For every topic there will be a managed ledger created.
  try {
    String path = String.format("/managed-ledgers/%s/persistent", namespaceName);
    if (LOG.isDebugEnabled()) {
      LOG.debug("Getting children from managed-ledgers now: {}", path);
    }
    for (String topic : pulsar.getLocalZkCacheService().managedLedgerListCache().get(path)) {
      topics.add(String.format("persistent://%s/%s", namespaceName, Codec.decode(topic)));
    }
  } catch (KeeperException.NoNodeException e) {
    // NoNode means there are no persistent topics for this namespace
  }
  topics.sort(null);
  return topics;
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

private Map<Long, Set<ResourceUnit>> getAvailableBrokers(ServiceUnitId serviceUnitId) throws Exception {
  Map<Long, Set<ResourceUnit>> availableBrokers = sortedRankings.get();
  // Normal case: we are the leader and we do have load reports information available
  if (availableBrokers.isEmpty()) {
    // Create a map with all available brokers with no load information
    Set<String> activeBrokers = availableActiveBrokers.get(LOADBALANCE_BROKERS_ROOT);
    List<String> brokersToShuffle = new ArrayList<>(activeBrokers);
    Collections.shuffle(brokersToShuffle);
    activeBrokers = new HashSet<>(brokersToShuffle);
    availableBrokers = Maps.newTreeMap();
    for (String broker : activeBrokers) {
      ResourceUnit resourceUnit = new SimpleResourceUnit(String.format("http://%s", broker),
          new PulsarResourceDescription());
      availableBrokers.computeIfAbsent(0L, key -> Sets.newTreeSet()).add(resourceUnit);
    }
    log.info("Choosing at random from broker list: [{}]", availableBrokers.values());
  }
  return availableBrokers;
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

protected Set<String> clusters() {
  try {
    Set<String> clusters = pulsar().getConfigurationCache().clustersListCache().get();
    // Remove "global" cluster from returned list
    clusters.remove(Constants.GLOBAL_CLUSTER);
    return clusters;
  } catch (Exception e) {
    throw new RestException(e);
  }
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

private void updateRanking() {
  try {
    synchronized (currentLoadReports) {
      currentLoadReports.clear();
      Set<String> activeBrokers = availableActiveBrokers.get();
      for (String broker : activeBrokers) {
        try {
          String key = String.format("%s/%s", LOADBALANCE_BROKERS_ROOT, broker);
          LoadReport lr = loadReportCacheZk.get(key)
              .orElseThrow(() -> new KeeperException.NoNodeException());
          ResourceUnit ru = new SimpleResourceUnit(String.format("http://%s", lr.getName()),
              fromLoadReport(lr));
          this.currentLoadReports.put(ru, lr);
        } catch (Exception e) {
          log.warn("Error reading load report from Cache for broker - [{}], [{}]", broker, e);
        }
      }
      updateRealtimeResourceQuota();
      doLoadRanking();
    }
  } catch (Exception e) {
    log.warn("Error reading active brokers list from zookeeper while re-ranking load reports [{}]", e);
  }
}

代码示例来源：origin: org.apache.pulsar/pulsar-broker

activeBrokers = availableActiveBrokers.get();

代码示例来源：origin: org.apache.pulsar/pulsar-broker

@GET
@Path("/{cluster}/failureDomains")
@ApiOperation(value = "Get the cluster failure domains", response = FailureDomain.class, responseContainer = "Map")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
public Map<String, FailureDomain> getFailureDomains(@PathParam("cluster") String cluster) throws Exception {
  validateSuperUserAccess();
  Map<String, FailureDomain> domains = Maps.newHashMap();
  try {
    final String failureDomainRootPath = pulsar().getConfigurationCache().CLUSTER_FAILURE_DOMAIN_ROOT;
    for (String domainName : failureDomainListCache().get()) {
      try {
        Optional<FailureDomain> domain = failureDomainCache()
            .get(joinPath(failureDomainRootPath, domainName));
        if (domain.isPresent()) {
          domains.put(domainName, domain.get());
        }
      } catch (Exception e) {
        log.warn("Failed to get domain {}", domainName, e);
      }
    }
  } catch (KeeperException.NoNodeException e) {
    log.warn("[{}] Failure-domain is not configured for cluster {}", clientAppId(), cluster, e);
    return Collections.emptyMap();
  } catch (Exception e) {
    log.error("[{}] Failed to get failure-domains for cluster {}", clientAppId(), cluster, e);
    throw new RestException(e);
  }
  return domains;
}

内容来源于网络，如有侵权，请联系作者删除！

ZooKeeperChildrenCache

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com