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

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

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

ZooKeeperCache介绍

[英]Per ZK client ZooKeeper cache supporting ZNode data and children list caches. A cache entry is identified, accessed and invalidated by the ZNode path. For the data cache, ZNode data parsing is done at request time with the given Deserializer argument.
[中]Per ZK client ZooKeeper缓存支持ZNode数据和子列表缓存。缓存条目由ZNode路径标识、访问和失效。对于数据缓存，ZNode数据解析在请求时使用给定的反序列化器参数完成。

代码示例

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

private void initZK() throws PulsarServerException {
  String[] paths = new String[] { MANAGED_LEDGER_ROOT, OWNER_INFO_ROOT, LOCAL_POLICIES_ROOT };
  // initialize the zk client with values
  try {
    ZooKeeper zk = cache.getZooKeeper();
    for (String path : paths) {
      if (cache.exists(path)) {
        continue;
      }
      try {
        ZkUtils.createFullPathOptimistic(zk, path, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      } catch (KeeperException.NodeExistsException e) {
        // Ok
      }
    }
  } catch (Exception e) {
    LOG.error(e.getMessage(), e);
    throw new PulsarServerException(e);
  }
}

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

&& exists(path, watcher)) {
  return getChildren(path, watcher);
} else if (cause instanceof KeeperException) {
  throw (KeeperException) cause;

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

@Override
public void reloadCache(final String path) {
  try {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Reloading ZooKeeperDataCache at path {}", path);
    }
    cache.invalidate(path);
    Optional<Entry<T, Stat>> cacheEntry = cache.getData(path, this, this);
    if (!cacheEntry.isPresent()) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Node [{}] does not exist", path);
      }
      return;
    }
    for (ZooKeeperCacheListener<T> listener : listeners) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Notifying listener {} at path {}", listener, path);
      }
      listener.onUpdate(path, cacheEntry.get().getKey(), cacheEntry.get().getValue());
      if (LOG.isDebugEnabled()) {
        LOG.debug("Notified listener {} at path {}", listener, path);
      }
    }
  } catch (Exception e) {
    LOG.warn("Reloading ZooKeeperDataCache failed at path: {}", path, e);
  }
}

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

public CompletableFuture<Optional<Entry<T, Stat>>> getWithStatAsync(String path) {
  return cache.getDataAsync(path, this, this).whenComplete((entry, ex) -> {
    if (ex != null) {
      cache.asyncInvalidate(path);
    }
  });
}

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

@Override
public void reloadCache(final String path) {
  try {
    cache.invalidate(path);
    Set<String> children = cache.getChildren(path, this);
    LOG.info("reloadCache called in zookeeperChildrenCache for path {}", path);
    for (ZooKeeperCacheListener<Set<String>> listener : listeners) {
      listener.onUpdate(path, children, null);
    }
  } catch (KeeperException.NoNodeException nne) {
    LOG.debug("Node [{}] does not exist", nne.getPath());
  } catch (Exception e) {
    LOG.warn("Reloading ZooKeeperDataCache failed at path:{}", path);
  }
}

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

public ZooKeeper getZooKeeper() {
  return this.cache.getZooKeeper();
}

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

public <T> Optional<Entry<T, Stat>> getEntry(final String path, final Deserializer<T> deserializer) throws Exception {
  return getData(path, this, deserializer);
}

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

public void asyncInvalidate(String path) {
  backgroundExecutor.execute(() -> invalidate(path));
}

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

/**
 * Simple ZooKeeperChildrenCache use this method to invalidate cache entry on watch event w/o automatic re-loading
 *
 * @param path
 * @return
 * @throws KeeperException
 * @throws InterruptedException
 */
public Set<String> getChildren(final String path) throws KeeperException, InterruptedException {
  return getChildren(path, this);
}

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

/**
 * Returns if the node at the given path exists in the cache
 *
 * @param path
 *            path of the node
 * @return true if node exists, false if it does not
 * @throws KeeperException
 * @throws InterruptedException
 */
public boolean exists(final String path) throws KeeperException, InterruptedException {
  return exists(path, this);
}

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

public void close() {
    if (this.rackawarePolicyZkCache.get() != null) {
      this.rackawarePolicyZkCache.get().stop();
    }
    if (this.clientIsolationZkCache.get() != null) {
      this.clientIsolationZkCache.get().stop();
    }
  }
}

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

protected static CompletableFuture<PartitionedTopicMetadata> fetchPartitionedTopicMetadataAsync(
    PulsarService pulsar, String path) {
  CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
  try {
    // gets the number of partitions from the zk cache
    pulsar.getGlobalZkCache().getDataAsync(path, new Deserializer<PartitionedTopicMetadata>() {
      @Override
      public PartitionedTopicMetadata deserialize(String key, byte[] content) throws Exception {
        return jsonMapper().readValue(content, PartitionedTopicMetadata.class);
      }
    }).thenAccept(metadata -> {
      // if the partitioned topic is not found in zk, then the topic is not partitioned
      if (metadata.isPresent()) {
        metadataFuture.complete(metadata.get());
      } else {
        metadataFuture.complete(new PartitionedTopicMetadata());
      }
    }).exceptionally(ex -> {
      metadataFuture.completeExceptionally(ex);
      return null;
    });
  } catch (Exception e) {
    metadataFuture.completeExceptionally(e);
  }
  return metadataFuture;
}

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

@POST
@Path("/racks-info/{bookie}")
@ApiOperation(value = "Updates the rack placement information for a specific bookie in the cluster")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
public void updateBookieRackInfo(@PathParam("bookie") String bookieAddress, @QueryParam("group") String group,
    BookieInfo bookieInfo) throws Exception {
  validateSuperUserAccess();
  if (group == null) {
    throw new RestException(Status.PRECONDITION_FAILED, "Bookie 'group' parameters is missing");
  }
  Optional<Entry<BookiesRackConfiguration, Stat>> entry = localZkCache()
      .getEntry(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, (key, content) -> ObjectMapperFactory
          .getThreadLocal().readValue(content, BookiesRackConfiguration.class));
  if (entry.isPresent()) {
    // Update the racks info
    BookiesRackConfiguration racks = entry.get().getKey();
    racks.updateBookie(group, bookieAddress, bookieInfo);
    localZk().setData(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper().writeValueAsBytes(racks),
        entry.get().getValue().getVersion());
    log.info("Updated rack mapping info for {}", bookieAddress);
  } else {
    // Creates the z-node with racks info
    BookiesRackConfiguration racks = new BookiesRackConfiguration();
    racks.updateBookie(group, bookieAddress, bookieInfo);
    zkCreate(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper().writeValueAsBytes(racks));
    log.info("Created rack mapping info and added {}", bookieAddress);
  }
}

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

private void saveQuotaToZnode(String zpath, ResourceQuota quota) throws Exception {
  ZooKeeper zk = this.localZkCache.getZooKeeper();
  if (zk.exists(zpath, false) == null) {
    try {
      ZkUtils.createFullPathOptimistic(zk, zpath, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    } catch (KeeperException.NodeExistsException e) {
    }
  }
  zk.setData(zpath, this.jsonMapper.writeValueAsBytes(quota), -1);
}

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

public CompletableFuture<Optional<T>> getAsync(String path) {
  CompletableFuture<Optional<T>> future = new CompletableFuture<>();
  cache.getDataAsync(path, this, this).thenAccept(entry -> {
    future.complete(entry.map(Entry::getKey));
  }).exceptionally(ex -> {
    cache.asyncInvalidate(path);
    future.completeExceptionally(ex);
    return null;
  });
  return future;
}

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

public Optional<Entry<T, Stat>> getWithStat(final String path) throws Exception {
  return cache.getData(path, this, this);
}

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

globalZkCache().invalidate(clusterPath);
  log.info("[{}] Updated cluster {}", clientAppId(), cluster);
} catch (KeeperException.NoNodeException e) {

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

public Set<String> get() throws KeeperException, InterruptedException {
  return cache.getChildren(path, this);
}

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

private void setDynamicConfigurationToZK(String zkPath, Map<String, String> settings) throws IOException {
  byte[] settingBytes = ObjectMapperFactory.getThreadLocal().writeValueAsBytes(settings);
  try {
    if (pulsar.getLocalZkCache().exists(zkPath)) {
      pulsar.getZkClient().setData(zkPath, settingBytes, -1);
    } else {
      ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), zkPath, settingBytes, Ids.OPEN_ACL_UNSAFE,
          CreateMode.PERSISTENT);
    }
  } catch (Exception e) {
    log.warn("Got exception when writing to ZooKeeper path [{}]:", zkPath, e);
  }
}

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

public void close() throws IOException {
  ZooKeeper currentSession = zkSession.getAndSet(null);
  if (currentSession != null) {
    try {
      currentSession.close();
    } catch (InterruptedException e) {
      throw new IOException(e);
    }
  }
  super.stop();
}

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

ZooKeeperCache

相关文章

    查看更多

    热门标签

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

    最新文章

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