问题标签 [pulsar]

我是脉冲星的新手！

现在，我正在尝试在 Scala中实现这些代码（ https://flink.apache.org/2019/05/03/pulsar-flink.html ）。但是，我找不到一些课程（例如PulsarSourceBuild）。我怎样才能在 Scala 中完成它们？我应该在哪里检查？

我有一个使用 Ignite 并安装了 Ignite 版本的 log4j2 的复杂项目。

我还想为 log4j2 安装一个Pulsar appender，但是，当我启动应用程序时，我收到以下错误：

显然，这两个依赖项之间存在冲突。但是，当我将 Pulsar appender 导入我的pom.xml时，我已经排除了大多数：

这就是我的log4j2.xml 的样子：

我知道您可以将主题订阅设置为共享订阅，以允许同一主题上的多个消费者。这也可以为多个生产者完成吗？

出于某种原因，当我尝试这样做时，我得到了一个Producer with name '<topic_name>' is already connected to topic

我有一个 Maven 依赖项 pulsar-log4j2-appender，我分叉并更改了源代码，因为它在我的项目中引发异常。

更改源代码后，我运行 mavenpackage命令构建 jar 并将其导入到我的项目中（在 Intellij 中：Project Structure | Modules | Dependencies | Add JARs or directories...).

但是，当我运行应用程序时，它似乎无法找到该依赖项，因为我在 log4j2.xml 文件中声明的 Pulsar appender 没有被配置。

我是否正确导入了 JAR？我想知道 JAR 是否需要在org.apache.pulsar命名空间内才能正确导入。

例如，

这是依赖项看起来没有改变的样子：

这就是我自己修改和构建它时的样子：

Pulsar 允许多个生产者订阅同一个主题，前提是它们具有不同的生产者名称。有没有办法检查具有相同名称（和相同主题）的生产者是否已经存在？

我认为一个简单的例子可以更好地描述我的问题。

例如，假设有一个名为“A”的主题，我已经生成了 100 条消息（message1...message100）。我已经使用独占类型的订阅“A_1”消费并确认了消息 50。由于某种原因，我的应用程序关闭了，所以在重新启动应用程序时，我需要再次读取消息 1。这可以实现吗？我在想有可能创建一个新订阅（“A_2”）并再次开始阅读消息，但我不确定“A_2”是否会从 message1 或 message51 开始阅读......任何方向或提示都会很棒！

提前致谢

我在这里按照手册操作：脉冲星功能

我在 Java 中有这个函数声明：

我建了这个罐子。将其复制到我的 pulsar stanalone docker 并运行以下命令：

进而：

现在我在 C# 中使用一些客户端来使用队列，当我阅读消息时，我看到“测试”而不是“测试！”。

所以似乎函数触发器有效，但函数本身没有生效。

我尝试在 dotnet 客户端中实现此 PR https://github.com/apache/pulsar/pull/8372 。据我目前所知，为输入类生成消息模式的工作方式如下：Some class -> .proto for this class -> google wrapper from this .proto -> get Descriptor from the wrapper 并继续进行序列化。

在集成测试中有一个例子可以做到这一点：

其中 TestMessage 是 .proto 定义：

我无法理解 .proto 是如何转换为 java 类的。有一个 proto 编译器https://developers.google.com/protocol-buffers/docs/reference/java-generated 可以从 .proto 创建 java 类；但它需要手动调用，然后才能将类包含在代码库中并使用。看起来它在集成测试示例中“即时”运行，至少我无法在任何地方找到生成的类。

C# 参考https://developers.google.com/protocol-buffers/docs/reference/csharp-generated做同样的事情，但生成 C# 源文件。

那么它究竟是如何在 java 客户端中工作的呢？

我无法获得一个简单的 hello-world 类型的 Spring Boot 应用程序来利用 Pulsar Log4j2 appender。在启动时它报告错误：

2021-05-14T15:34:23.549-05:00 [pulsar-client-io-6-1] 错误 oapciProducerImpl [applog-json] [pulsar-log4j2-appender-applog-json] 无法创建生产者：生产者名称'pulsar-log4j2-appender-applog-json' 已经连接到主题 2021-05-14 15:34:23,550 主要错误无法启动 pulsar manager org.apache.pulsar.client.api.PulsarClientException$ProducerBusyException：Producer with name 'pulsar-log4j2-appender-applog-json' 已经连接到 org.apache.pulsar.client.impl.ProducerBuilderImpl 的 org.apache.pulsar.client.api.PulsarClientException.unwrap(PulsarClientException.java:884) 的主题。在 org.apache.pulsar.log4j2.appender.PulsarManager.startup(PulsarManager.java:127) 在 org.apache.pulsar.log4j2.appender.PulsarAppender.start(PulsarAppender.java:187) 处创建(ProducerBuilderImpl.java:93) ...

我正在使用 Apache Pulsar 2.7.1 和 pulsar-log4j2-appender:2.6.3 （我尝试了其他版本的 appender，结果相同）。这与错误可能暗示的运行多个实例无关。在启动应用程序之前，我什至可以通过“pulsar-admin topics stats”验证当前没有连接到该主题的发布者。我使用 Pulsar Log4j2 appender 启动了一个 Spring Boot 应用程序实例，它会记录该错误并且不会向 Pulsar 发送数据。如果错误消息确实有效，则附加程序必须以某种方式在内部创建多个连接。有没有其他人遇到过这个并想出一个解决方案？这似乎是一件相当简单的尝试和做的事情，不应该如此复杂或挑剔。

我将示例项目发布到 GitHub，以便有人更轻松地对其进行测试。

编辑 pulsar.host 变量以反映 Pulsar 实例的主机名。这假定它位于 http://localhost:6650。这在 log4j2.xml 文件中被引用。当我执行此操作时，我得到了提到的错误。
注意我创建了另一个不使用 Spring Boot 的示例 Java 项目，它工作正常。它似乎与 Spring 依赖项有关。

我使用 pulsar perf 通过以下命令测试服务：pulsar-perf generate topic1 -n 1 -t 10 -r 1000000，但是日志显示Throughtput Produced 刚好低于 10000，我需要通过更高的消息率测试服务， 我能怎么做 ？