问题标签 [apache-pulsar]

我的团队最近在考虑为我们的项目使用不同的消息代理，我们最终选择了 Apache Pulsar，但它适用于其他人（Kafka）。我们的要求是跟踪发送给每个订阅者的消息总数和发送字节数以进行计费。

我正在阅读有关指标的文档，并惊讶地发现 Pulsar 没有跟踪这一点，我检查了 Kafka，结果是一样的。我对这个主题的理解很少，所以这是某种反模式吗？我知道像这样的计数器值永远不会下降，对于我们的用例 - 不应该重置，导致潜在的（某些）溢出。但对我来说，这可以通过在 Prometheus 中使用直方图（Pulsar 中使用的度量格式）来解决。我实际上正在考虑实现这样的功能，但我错了，是否有更好的解决方案来满足我们的目的？

我正在构建一个处理 Apache Pulsar 的 GO 应用程序。正如 Pulsar 文档所要求的那样，Go 客户端需要 C++ 库（顺便说一句，Kafka 也是如此）。

我想将所有这些打包成一个容器，尽可能最小。我通常使用 SCRATCH 并从另一个基于 golang 的容器中复制输出。不幸的是，我无法从这个初始容器中获取外部库：

使用它会使构建失败，寻找丢失的符号

而本地构建工作。

如何正确集成我需要的库？

我正在使用 Flink 从 Apache Pulsar 读取数据。我在 pulsar 中有一个分区主题，有 8 个分区。我在这个主题中生成了 1000 条消息，分布在 8 个分区中。我的笔记本电脑有 8 个内核，所以我有 8 个子任务（默认并行度 = 内核数）。我从 Eclipse 执行代码后打开 Flink-UI，发现一些子任务没有收到任何记录（空闲）。我期望所有 8 个子任务都将被利用（我期望每个子任务都将映射到我的主题中的一个分区）。

重新启动作业后，我发现有时使用了 3 个子任务，有时使用了 4 个任务，而其余的子任务保持空闲。

请您的支持澄清这种情况。

另外我怎么知道子镜头之间是否有洗牌？

我的代码：

https://pulsar.apache.org/docs/en/next/admin-api-persistent-topics/

https://pulsar.apache.org/docs/en/next/admin-api-non-persistent-topics/

我有一个主题，对于大多数订阅者来说，持久性很重要。我想（暂时）使用一个在断开连接后不关心消息的客户端订阅该主题，因此我不希望将大量消息写入其积压。据我所知，持久性只能在主题级别上进行配置。有没有办法实现这种基于订阅者的持久性？

（换句话说，与使用 cleanSession=True 连接到 MQTT 的等效功能）

我正在尝试在 ubuntu 上为 node js 客户端安装 pulsar-client。我收到以下错误：

文件说

Pulsar Node.js 客户端库基于 C++ 客户端库。按照 C++ 库的说明通过 RPM、Deb 或 Homebrew 包安装二进制文件。

（注意：您不仅需要安装 pulsar-client 库，还需要安装 pulsar-client-dev 库）

但是，在 Ubuntu 上安装 C++ 库的说明不是很清楚。也不清楚如何安装 pulsar-client-dev 库。

我正在尝试使用 Apache Pulsar 编写一个多租户（多个命名空间）系统。我似乎无法实现的是使用单个客户端/连接发布到这些名称空间中的多个名称空间和主题。消费者方面也是如此。本质上，我想使用单个应用程序实例来发布到多个命名空间和主题，而消费者应用程序能够在消费时做到这一点。我只是想了解这是否可能。

我不想为 M 命名空间和每个命名空间的 N 个主题维护 M*N 连接。我看到 pulsar 支持正则表达式，但是如果我想分别使用每个主题，这在消费者方面意味着什么（我是否会从所有匹配正则表达式的主题中获得混合消息？）

假设有一个 Pulsar Producer 用于持久主题topic1（命名空间和租户与问题无关）。

假设我们有多个消费者为同一个主题 ( topic1) 使用不同的订阅名称。

是否可以配置消费者接收相同的消息？因此，例如，如果将消息msg1发送到主题consumer1并consumer2接收此消息？

消费者和生产者都是用 Java 编写的，但编程语言并不重要。

澄清

当前对具有多个订阅的同一主题的多个订阅观察到的行为是，每个订阅者都不会收到已发布到该主题的所有消息。我需要接收来自该主题的所有消息。

我得到了这个例外：

当尝试从我们的客户端应用程序连接到 Pulsar 时。我正在运行 Pulsar 2.4.2。

我确认我正在连接到正确的端点 ( pulsar+ssl://pulsar-ms-tls.mydomain.com:6651)，并且我们正在使用 SSL+TLS。

什么可能导致这个问题？

我需要有关 Pulsar 函数异常的帮助，什么是用户函数异常，如果它的速率/计数很高会发生什么，以及高速率/计数的原因。

当 Producer 设置为 customPartition 时，无法从生产者向 pulsar 发送消息（请参阅下面的代码）。

发送消息的代码：

MessageRouterImpl 随机生成范围从 0 到 5 的数字，如下代码

}

我的问题是为什么我无法使用CustomPartition从生产者那里发送消息以及为什么我收到低于日志消息的消息

从定时器线程批处理来自批处理容器的消息

使用 0 条消息对来自批处理容器的消息进行批处理

使用 MessageRoutingMode.RoundRobinPartition 和 MessageRoutingMode.SinglePartition 我能够从生产者那里发送消息。

如果有人对此有所了解，那将非常有帮助。