org.apache.pulsar.zookeeper.ZooKeeperChildrenCache.()方法的使用及代码示例

转载 作者：知者 更新时间：2024-03-17 18:23:31 26 4

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

/**
 * 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());
  }
}

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

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

availableActiveBrokers = new ZooKeeperChildrenCache(pulsar.getLocalZkCache(), LOADBALANCE_BROKERS_ROOT);
availableActiveBrokers.registerListener(new ZooKeeperCacheListener<Set<String>>() {
  @Override

availableActiveBrokers = new ZooKeeperChildrenCache(pulsar.getLocalZkCache(),
    LoadManager.LOADBALANCE_BROKERS_ROOT);
availableActiveBrokers.registerListener(new ZooKeeperCacheListener<Set<String>>() {

26 4 0

• apache-pulsar - Pulsar 消息的消息过期

  如何使用 Pulsar 管理 API(或)获取 Pulsar 中特定消息的到期日期？有什么方法可以查明特定消息是否已过期？ 最佳答案 https://pulsar.apache.org/docs/en

• apache-pulsar - 在 Pulsar 中排除发件人的主题消费者

  简介 我们正在开发一个系统来支持多个实时消息(聊天)和更新(事件通知)。 也就是说，用户 A 可以通过 Web Socket 接收消息: 接收新的聊天消息 接收某些事件的更新，例如，如果有人喜欢他们的

• apache-pulsar - 有没有办法检查是否存在关于同一主题的同名 Pulsar 生产者？

  Pulsar 允许多个生产者订阅同一个主题，前提是他们的生产者名称不同。有没有办法检查是否已经存在具有相同名称(和相同主题)的生产者？ 最佳答案 您可以使用 stats来自 pulsar-admin

• apache-pulsar - Apache pulsar 无限保留

  在 Apache Pulsar 主题文档中，它说我们可以将主题时间保留策略设置为 -1 以实现基于无限时间的保留，无限保留的缺点是什么，我们可以使用 pulsar 作为消息存储，其中数据永远存在于主题

• apache-pulsar - 用 pulsar 创建很多主题是一个好场景吗？

  使用 pulsar 为每个散列动态创建一个主题(或分区)，并在不再使用时删除主题(或分区)，这是可能的和好的场景吗？ 这个想法是能够以有序的方式读取具有相同散列的数据，而无需在具有相同散列的两条消息之

• apache-pulsar - 从 apache pulsar 主题中检索所有过去的消息

  我认为一个简单的例子可以更好地描述我的问题。 例如，假设有一个名为“A”的主题，我已经生成了 100 条消息 (message1...message100)。我已经使用独占类型的订阅“A_1”使用并确

• apache-pulsar - 如何缩小 pulsar BookKeeper 集群

  问题 一个月前我向 pulsar 添加了 2 个 bookie，但我意识到这是浪费资源。我怎样才能将博彩公司从 7 个减少到 3 个。 我的努力 通过 bookkeeper 的文档，我发现当 book

• apache-pulsar - 使用 Apache Pulsar 中注册的模式发布到主题

  如 Pulsar Schema Registry Docs 中的示例所示 Producer producer = client.newProducer(JSONSchema.of(User.class

• apache-pulsar - 如何缩小 pulsar BookKeeper 集群

  问题 一个月前我向 pulsar 添加了 2 个 bookie，但我意识到这是浪费资源。我怎样才能将博彩公司从 7 个减少到 3 个。 我的努力 通过 bookkeeper 的文档，我发现当 book

• apache-pulsar - 使用 Apache Pulsar 中注册的模式发布到主题

  如 Pulsar Schema Registry Docs 中的示例所示 Producer producer = client.newProducer(JSONSchema.of(User.class

• apache-pulsar - 删除/过期 Apache Pulsar 主题中所有消息的最有效方法是什么？

  我正在尝试找出从 Pulsar 主题中删除所有消息(逻辑上或物理上)的最佳方法，以便订阅不再使用它们？ 我知道我们可以简单地执行 $ pulsar-admin persistent delete pe

• apache-pulsar - Apache Pulsar Java 客户端占用过多内存 (OOM)

  我用 Spring boot 编写了一个简单的 Apache Pulsar 客户端——一个初始化为 bean 的 pulsar-producer，它将在 rest controller 中用于将传入的

• apache-pulsar - Apache Pulsar 未经授权验证关于主题的操作 [WRITE] 的 TopicPolicyOperation

  我们一直在尝试在我们的 2.8.1 Apache Pulsar 集群上添加 JWT 身份验证，它似乎工作正常，但权限除外。所以基本上，我们现在的状态是: 没有 token ，客户端会收到 401 使用

• java - Pulsar 消息监听器的多线程

  我对 java 消息监听器 和 apache pulsar 还很陌生。假设我已经维护了一个这样的监听器， private MessageListener generateListener() {

• apache-pulsar - 非持久消息在吞吐量高时丢失

  我发现即使我的 Pulsar 客户端已启动并正在运行，非持久性消息有时也会丢失。那些非持久化的消息在吞吐量很高(很短的时间内超过1000条消息。我个人认为这并不高)时丢失。如果我增加参数 receiv

• 八张图带你彻底理解 Pulsar 的跨地域复制

  跨地域复制是 Apache Pulsar 企业级特性的重要组成部分，它保证了系统的高可用，在操作和管理上也非常便捷，今天用 5 张图来带大家学习这个功能。 1.多机房部署 Pulsar 多机

• python - 无法安装 pulsar-client

  我根本无法安装文档中提到的 pulsar-client: pip3 install pulsar-client Collecting pulsar-client ERROR: Could not fi

• stream - Apache Pulsar 支持的最大消息大小

  Apache Pulsar 支持的最大消息大小是多少？ 我尝试阅读文档。但我找不到任何相关信息。 最佳答案 默认的最大消息大小为 5242880 字节 (5 MiB)，但可以调整。有关详细信息，请参阅

• docker - Kubernetes集群上的Apache pulsar(本地环境)

  尝试在我的kubernetes集群中部署P​​od和一些Pod给了我一些存储问题的错误。屏幕截图如下: 我确定问题出在我的一个工作节点上。我认为 pulsar 不是问题。我还将在这里共享YAML文件，

• python - 如何更改运行 Pulsar 的端口？

  我想将 pulsar 与 apache airflow 一起运行。问题是两者都在端口号 8080 上运行。我不想更改 Airflow 的配置，但为了让 pulsar 运行，我必须为其分配另一个端口。我