问题标签 [kafka-node]

我创建了单个连接器以将集合的所有插入/更新事件推送到同一主题（1 个分区）。处理插入事件消息的消费者代码将比更新事件花费一些时间。

这里的问题是消息没有根据消费顺序提交。以下是Kafka消费者的步骤和行为。

1. 在集合中插入一条记录并在一秒钟内更新该字段。
2. 该主题现在有 2 条记录。一个用于插入事件（消息 1），另一个用于更新事件（消息 2）。
3. 消息 1 被消费
4. 消息 2 被消费
5. 消息 2 已提交
6. 消息 1 已提交

有什么办法可以等到消息 1 提交后再使用消息 2。

消费者守则：

所以我有一个消费者组正在运行并收听来自某个主题的消息。如何将更多的消费者实例添加到消费者组？如果这就是我收听消息的方式

我尝试创建消费者并在选项中包含相同的组名，

但是当我运行命令时，我看不到它反映在消费者组中

我看到同一个消费者被分配了不同的主题分区。

如何将更多消费者添加到已创建的现有消费者组中？

nodejs 服务器使用 pm2 作为一个实例分支启动。

我的卡夫卡选项：

客户：

制片人：

消费者：

有什么建议为什么它不断发生？

我正在尝试在 AWS 中创建一个 Lambda，作为 MSK 主题的生产者。所有 AWS 文档都说要创建一个新的 EC2 实例，但由于我的 Lambda 在同一个 VPC 中，我觉得这应该可以工作。我对此很陌生，我注意到我的日志语句从未在我的 producer.on 函数中命中。我正在使用 nodejs 和 kafka-node 模块。代码可以在下面找到。

本质上，我只是想知道是否有人知道如何做到这一点，以及为什么当我通过 Lambda 运行测试时，producer.on 函数永远不会被命中？这只是一些测试代码，看看我是否可以发送它，但如果需要更多数据以提供帮助，请提前告诉我并感谢。

我的 nodejs 代码的节点。我有一个用于请求/响应的 api。首先，我发出一个请求http://localhost:3000/number1，在我启动一个消费者之后，它使用来自 kafka 主题和一个分区“接收”的消息，然后我尝试找到 id = 的消息number1。在我想用这个值向用户返回一个响应之后。所以我创建了一个消费者，如下所示：

如果我尝试提出一个连续的请求，在第一个请求之后我会得到一个响应：

我的 Docker-compose file1 如下：

码头工人撰写文件2

为什么会这样？你能帮我解决这个问题吗？

[如果您想了解更多信息，请随时问我:)]

我在 nodejs 中使用 kafka-node 模块。我想确定主题是否存在。

var kafka = require('kafka-node'); 客户端 = 新 kafka.KafkaClient(),

我很想在节点上使用 kafka，并试图找出最好的库是什么。

我从 kafka-node 开始，但看起来 node-rdkafka 或 kafkajs 更高级并且正在维护。我正在做kafka的基本用法，发布到1个主题，消费消息并在代码中处理它。我需要能够控制消费者的速度（使用 kafka-node 我必须使用 quque 来存储消息，暂停消费者并在每个队列批处理后恢复）有人有最新的推荐或比较吗？

谢谢

我已经用主题'topic1'（在模式注册表中使用模式）测试了 Kafka 集群。

我从这个主题创建了流：

然后我创建了聚合表作为选择推送查询（没有窗口），只有今天的数据：

从 ksqlDB CLI 我运行并看到 OK 结果：

我也可以topic1_agg_table_JSON在主题列表中看到，里面有 json 消息。

目标：我想在 node.js 中编写使用者以将这些消息（使用 websockets）发送到浏览器（客户端）并在客户端（实时）可视化它。

已经尝试过： kafka-node模块。示例代码取自https://github.com/SOHU-Co/kafka-node/blob/master/example/consumer.js使用简单的原始（简单）Kafka 主题，在这段代码中一切正常，但如果我会改变我的主题topic1_agg_table_JSON，它不会抛出任何错误，也不会打印任何消息。

问题：topic1_agg_table_JSON使用 Node.js使用数据的正确方法是什么？

请原谅我对服务器和防火墙的有限知识。

我目前经常运行一个 Kafka（nodejs 实现，node-rdkafka）生产者脚本来提供分析。这一切都通过 Docker 容器运行良好。

一旦进入 AWS (EC2)，出站请求就会被本地防火墙阻止。

我们的 devops 部门已经声明他们已经开放了一个端口来允许这个请求（比如说 8765）。

我的问题是，如何运行节点生产者脚本以使用该端口对 Kafka 服务器的出站请求？

我正在尝试使用 kafka-node 将我们的应用程序与 Kafka 集成。我从插件站点获取示例代码并尝试对其进行模块化。我在后台使用融合的 Kafka docker 图像。

示例代码：

另一方面，使用以下代码访问此

当消息被写入 Kafka 主题时，一切似乎都运行良好，唯一的问题是一段时间后，我收到以下错误

我完全不知所措，因为在尝试解决这个问题时我对 nodejs 的了解已经完全耗尽（顺便说一句，这并不多）。