问题标签 [kafka-node]

在一个项目中，Node.js 应用程序连接到 Kafka 消息队列并从队列中获取所有消息。我搜索并找到了两个包kafka-node和kafka-streams。哪个包适合这里？

https://www.npmjs.com/package/kafka-node

https://www.npmjs.com/package/kafka-streams

我的 docker-compose 文件

生产者代码是

在向主题发送有效负载之前创建主题时出现以下错误

在调用createtopic方法之前，我已经延迟了 5 秒，但是没有找到运气。

请协助。

为什么我得到错误kafka.Client不是构造函数，同时npm kafka-node这是创建kafka连接的方式？

具体有问题的行是：this.client = new kafka.Client();
整个构造函数是：

我正在尝试使用kafka-node-avro

库将数据推送到现有主题。我已经使用 curl 和主题将模式添加到 SchemaRegistry 中。我收到以下错误：

我的代码片段如下：

我是 kafka 的新手，并使用 kafka-node 在 nodeJS 中实现它。我想在一个主题中创建 3 个分区并同时向所有主题发布消息。我尝试了以下代码，但这里只有一个分区正在创建，所有消息都将发送到该分区。谁能告诉我哪里出错了。太感谢了。

我收到如下回复-

我有单个 kafka 代理，并且正在使用 kafka-node 在 nodeJS 中实现 kafka。我想创建一个包含 3 个分区的主题。这样做时，出现的问题是只有第一个分区正在分配领导者，而其他两个分区没有获得领导者。我想为所有分区分配领导者。谁能告诉我我该怎么做？提前致谢。

主题创建如下 -

所以我的问题可能涉及基于应用程序性质的一些头脑风暴。

我有一个向 Kafka 发送消息的 Node JS 应用程序。例如，每次用户单击页面时，Kafka 应用程序都会根据访问情况运行计算。然后，我想在通过我的 Kafka 消息触发它之后检索这个计算。到目前为止，此计算存储在 Cassandra 数据库中。问题是，如果我们在计算完成之前尝试从 Cassandra 读取数据，那么我们将不会从数据库中查询任何内容（尚未插入密钥）并且不会返回任何内容（错误），或者计算可能是陈旧的。到目前为止，这是我的代码。

首先，我想到了 async 和 await ，但这被排除在外，因为这不会停止过时的计算。

其次，我考虑让我的应用程序休眠，但似乎这种方式会减慢我的应用程序速度。

我可能决定使用 Kafka Consumer（在我的 node-js 中）来使用一条消息，表明现在查看 Cassandra 表是安全的。

例如（使用 kafka-node）

这种方法虽然更好，但似乎有点不对劲，因为每次用户点击页面时我都必须让消费者成为消费者，而且我只关心它被发送 1 条消息。

我想知道我应该如何应对这个挑战？我是否可能错过了一个场景，或者有没有办法使用 kafka-node 来解决这个问题？我也在考虑做一个等待承诺成功并且计算不会过时的while循环（比较缓存中的值）

stackoverflow 本身有几个类似的问题，但我仍然要再次询问，因为我想要NodeJS中的解决方案。

情景是——

（我正在使用 kafkaJS 库）

我有两个主题 topic-A 和 topic-B，所以我订阅/topic-.*/，然后创建 topic-C，现在我希望我的消费者也自动订阅 topic-c。但这并没有发生。

我只得到一个解决方案，即我必须在每个新主题创建后停止我的消费者并重新启动它。（但是当多个主题将动态创建时，这种方法似乎很难维护）。

任何人都可以通过示例向我建议任何其他解决方案。先感谢您。

我目前正在使用 Nodejs 和 Kafka，我正在寻找一种方法来检测主题中分区数量的变化（目前正在使用kafka-node）是否有任何 API 可以实现此功能？

我想Client.loadMetadataForTopics()每隔一小时左右打电话来检查分区号，但想知道是否有其他更好的方法可用。也欢迎任何建议或文档链接。十分感谢！

最近我正在使用 Kafka-node 模块在 nodeJS 下编写一些代码，我想控制每个消费的最大轮询数。从大多数博客中，我可以使用“max.poll.records”在 java 中找到解决方案，但很难找到 nodeJS 的解决方案。

任何人都可以帮忙吗？