问题标签 [kafkajs]

我一直在研究 kafkajs 和 socket.io 我对它很陌生，我似乎无法理解一些事情。我创建了一个聊天应用程序，基本上通过打开浏览器（客户端），您可以键入消息并将它们显示在聊天窗口中。我找到了一个让 kafka 打印“this message + i”的教程。我不想发送到主题并打印 message+i 来打印人们在聊天中输入的内容，但我不确定我应该如何做到这一点。

这是我的 consumer.js：

这是我的 producer.js：

我知道我应该以某种方式将主题代理和客户端与 socket.io 连接，但我不确定如何。

这是我的chat.js：

我有一个 ELKK docker compose stack，我正在尝试将一个 kafka 客户端（kafka js）连接到 kafka，但没有运气。

码头工人-compose.yml

消费者.js

我得到的错误如下。

例如，我也尝试过网络

已分配给所有服务，但没有任何改变。

我也尝试过hostname: kafka to kafka service，但也没有任何改变。

但是，我可以使用localhost:29092从外部客户端成功连接到代理。

那么，我做错了什么？

我正在使用NestJS 适配器实现 kafka 适配器。在这个适配器中，我需要能够发布和订阅 Kafka 主题。适配器类不是控制器，所以我不能在那里添加 @EventPattern() 方法。我可以注入 kafkaClient 并且只能发送消息：

但似乎我无法使用这个 kafkaClient订阅Kafka 主题。消费者已创建但未暴露给公共 KafkaClient 接口。所以我被迫初始化另一个自定义消费者，它正在创建一个到 Kafka 的单独连接，我不喜欢它。

有什么方法可以在控制器之外订阅 Kafka 主题？

我正在使用 KAFKAJS 与 Kafka 消费者合作。我需要知道分配给特定消费者的所有分区。

例如，假设主题 1 有 5 个分区。并且有 2 个消费者具有相同的 clientId。因此，将为一个分配 3 个主题，另一个分配 2 个主题。我希望每个消费者都知道分配的分区。

我已经阅读了 nestjs 微服务和 kafka 文档，但我无法弄清楚其中的一些。如果你能帮助我，我将非常感激。因此，正如文档所说，我必须在 main.ts 文件中创建一个微服务，如下所示：

然后有一个像这样的kafkaModule文件：

首先我想不通的是 createMicroservice 的第一个参数有什么用？（我通过了 AppModule 和 KafkaModule 并且都正常工作。知道 kafkaModule 是在 appModule 中导入的）

另一件事是，据我了解，main.ts文件中的微服务部分和配置用于订阅MessagePattern或EventPattern装饰器中使用的主题，而kafkaModule中描述的kafkaClient用于发送消息到不同的话题。

这里的问题是，如果我之前说的是真的，那么为什么 clientModule 使用默认的 groupId 如果没有指定作为消费者工作。奇怪的是我找不到使用 clientModule 从任何主题获取任何消息的解决方案。我现在正在做的是在每个文件中使用不同的组 ID，这样它们就不会发生任何冲突。

我有这个非常简单的代码，它应该只是使用来自官方 GitHub的本地主机中的 Kafka 将一条简单的消息从生产者传输到消费者

当我使用node index.js（即此文件）执行此代码时，我得到了这两个错误中的许多：

所以看起来我猜在本地端口 9092 上定义代理的第一行不起作用？但我不知道如何纠正这一点。请问你能帮帮我吗？

所以我是 kafka 的新手，我很难找到有关拥有多个 kafka 实例的信息，以便在代理实例出现故障时让我的消息服务仍然正常运行。

我用 Kafka JS 做了一个小本地演示，我看到我们用一个像这样的数组声明我们的代理

所以我的问题是如何让我的 kafka 代理的 X 个实例通过消息同步但使用不同的存储/实例/服务器（不确定这里的合适词是什么）？

非常感谢 ！凯夫

我使用 kafkajs 公开从 kafka 主题读取的数据，以便通过 http 端点公开，以便普罗米修斯抓取数据。但我无法公开来自 kafka 主题的数据。我写过这样的生产者和消费者

生产者.js

索引.js

下面是消费者 Consumer.js

当我点击 api 时，我无法将使用的消息发送到 API

我正在使用 KafkaJS 创建主题。

出现以下错误：

message":"[Connection] Response CreateTopics(key: 19, version: 3)","broker":"HOST_NAME:9092","clientId":"example-producer","error":"请求参数不满足配置的策略","correlationId":3,"size":58}

但是，当我尝试列出集群中存在的主题时，我得到了成功的响应，因为主题已经存在。

我有发出多个版本记录的 kafka 主题。在我的解决方案中，我只想过滤掉非常具体的消息，例如

一些消息如下所示：

而其他人看起来像这样：

订阅消息流并注销整个消息时，我可以在控制台中看到 Avro 类型，但是在使用 message.value 时，该信息会丢失。我最终希望在 allowMessageFilter() 中有一些过滤逻辑来确定允许哪些记录。在这种情况下，我只想过滤“Memgroup_composite_2”

任何帮助将不胜感激。