我想使用Kafka绑定的testcontainers来集成测试我的应用程序。在下面的场景中，测试设置永远不会通过waitForMessagesToBeDelivered，它将永远等待在收到消息后创建实体。在我的理解中，它应该通过Kafka传递两个消息，这两个消息应该被MyConsumer接收，然后应该创建两个MyEntity，这应该会增加myEntityRepository.count()返回的值。在我为其创建此再现
浏览 310提问于2019-11-16得票数 1
回答已采纳
1回答
我正在学习kafka，我想把我的应用程序分成两个微服务。首先，将所有传入的消息从KafkaConsumer保存到数据库，然后根据给定的id选择entity。第二，提供REST来保存和获取实体。他们之间的互动提供了卡夫卡。如何使用kafka从REST中的db接收存储的ID？这是在POST请求时调用生产者的示例代码。String, byte[]> result = resultFuture.get
浏览 2提问于2017-11-28得票数 3
回答已采纳
我正在尝试设计一个使用Kafka的处理代理来并行处理来自Kafka主题的消息。
如果一条消息被发布到某个主题，它是只存在于主题中的所有分区之间，还是在可能的多个分区上复制？我读过可以支持这两种可能性的声明。当我启动一个新的
浏览 7提问于2017-05-17得票数 0
我想使用kafka作为collectd的传输层。我发现有一个用于write_kafka的collectd插件，它将收集到的所有指标发送到kafka主题。我的意图是让一些主机作为收集器(作为kafka使用者组工作)，将这些指标从主题中删除，并将它们放入时间序列数据库(influxdb或graphite)。
collectd没有kafka使用者输入插
浏览 0提问于2017-02-11得票数 3
1回答
我使用spring-integration-kafka实现了使用自定义确认机制的Kafka使用者。
使用了来自的代码。我试图实现的是，当抛出异常时，不应该将确认发送回Kafka (即不执行偏移提交)，因此下一个fromKafka.receive(10000)方法调用将返回与前一个方法相同的消息。但我面临一个问题:即使确认没有发送给Kafka，消费
浏览 3提问于2015-10-21得票数 5
1回答
比方说，我有一个服务，它通过kafka-rest代理来消费消息，并且总是在同一个消费者群体上。我们还可以说，它正在消耗一个有一个分区的主题。当服务启动时，它在kafka-rest代理中创建一个新的使用者，并使用生成的使用者url直到服务被关闭。因此，在kafka-rest-proxy重新启动我的服务之后，现在就有了'n‘消费者，但他们中只有一个在积极消费。我甚至能够在我的
浏览 1提问于2016-07-07得票数 2
回答已采纳
我一直读到Kafka 只为单个分区分配一个使用者实例(线程)。最近我遇到了ConcurrentKafkaListenerContainerFactory，并且正在阅读它，有些方面似乎与我以前所知道的相矛盾。假设我们有一个使用者实例(pod/)，其中我使用的是并发级别设置为3 3的。我有一个有三个分区的Kafka主题。当并发级别设置为3时，我理解我的@KafkaListener方法将从主题的
浏览 6提问于2021-08-29得票数 1
回答已采纳
从理论上讲，既然nodejs是单线程的，那么当我定义多个使用者以提高吞吐量时，如何才能实现并行呢？例如，如果我有一个有4分区的卡夫卡主题，那么当与nodejs一起使用时，我如何能够并行地使用4条消息。最多我可以使用单线程事件循环来实现并发。一种可能的解决方案是分叉子进程(在本例中为3)，以便每个进程能够接收来自特定分区的消息，假设系统有3个空闲内核。但是，这种方法的效率/效力如何？
实现这一目标的最佳途径是
浏览 2提问于2020-05-18得票数 2
1回答
我正在进行一个项目，在这个项目中，生产者可以在运行时(而不是固定数量)出现并发布消息。目前，他们发布到在卡夫卡代理中运行时创建的唯一主题(topic.uuid)，我在另一端有一个用户，它订阅了主题。*模式和订阅所有主题，并且随着新主题的到来而重新平衡in.Is它是正确的方法吗？现在，我感到困惑的是，我们应该有一个具有多个分区的主题，还是有一个分区each.Technically的多个主题，这是相同的。但是，获得新分区(在运行时)和为每个分区(在运行时)
浏览 1提问于2016-11-19得票数 6
我们正在使用Kafka 0.9版本，并且有大量消息被推送到kafka主题中的特定分区。在本主题中有多个这样的分区。我们在本主题中为每个分区分配了一个使用者，并且在主题内部手动维护偏移量--外部数据存储中的分区。我想知道，如果我们开始在一个主题分区中获得大量的消息，那么是否有可能让使用者处理一个主题-分区为多线程。因为分配给分区的
浏览 2提问于2016-01-20得票数 3
2回答
我已经在docker上安装了confluent kafka。在主题中，我有10个分区。问题是我不能使用该主题中的消息，但我可以在该主题中生成消息。我正在尝试使用C# confluent.kafka驱动程序1.5.1 (最新)和librd.kafka 1.5.0 (最新)这一主题中的内容。我用来启动kafka的docker-com
浏览 51提问于2020-08-23得票数 2
回答已采纳
我有以下带有kafka输入的logstash配置 kafka { topics => ["codec => "json" doc_as_upsert => true }我面临的</e
浏览 5提问于2017-02-04得票数 2
在以下步骤中需要帮助:如何使用Kafka主题中的消息并将它们存储在目录/tmp/kakfa-messages中创建一个kafka使用者来使用来自主题“messages应用程序”的消息，并将它们存储在'/tmp/kafka-messages‘中
步骤1:我可以使用发布到主题“messages应用程序”的
浏览 0提问于2020-08-17得票数 0
回答已采纳
我在我的Kafka集群中有一个简单的观察(Kafka 0.11.0.0)。
我如何检查Kafka经纪人在做什么并自己产生这样的负载？
浏览 3提问于2017-11-10得票数 3
回答已采纳
我正在阅读卡夫卡，这是一本权威的指南，我看到了以下关于消费者的观点。
heartbeat.interval.ms必须低于session.timeout.ms，并且通常设置为超时值的三分之一。我可以知道为什么是这样的，我认为两者应该是相同的，比方说，如果我将session.timeout.ms设置为10，将心跳设置为3s，那么如果使用者需要更长的时间来处理记录，那么代理已经发送了调查结果，并计算出消费者已经死亡并重新平衡，在这种情况下，会话超时属性的目的
浏览 0提问于2019-12-09得票数 5
3回答
我正在为我们的应用程序实现一个基于Kafka的解决方案。根据Kafka文档，我所理解的是消费者组中的一个消费者(这是一个线程)内部映射到订阅主题中的一个分区。假设我有一个有40个分区的主题，并且有一个高级别的使用者在4个实例中运行。我不希望一个实例使用另一个实例使用的相同消息。但是，如果一个实例发生故障，则其他三个实例应该能够处
浏览 1提问于2017-06-16得票数 9
回答已采纳
现在，我们正在引入一个新的使用者组(称为k组)，它应该使用来自相同现有主题的数据，并且应该写入数据库。作为第一步，在开始使用现有主题中的消息之前，我们有一个初始数据迁移工作流，以获取源数据库的转储并将转储复制到目标数据库。
现在，当消费者群体开始时，我想知道它应该从什么开始？一种选择是使用latest。但问题是，当为这个新的使用者组执行初始数据迁
浏览 0提问于2020-10-07得票数 0
只是想知道在python脚本运行时是否可以重用使用者？ 我正在使用heroku kafka进行数据流，并且有一个python脚本，它使用kafka-python库连接到代理。运行该脚本时，将创建一个新的使用者组。然而，Heroku每天至少重启我的脚本一次，我无法停止重启。因此，每次重新启动脚本时，都会创建一个新的使用者，并且我会遇到重新平衡问题。只是想知道如何在脚本中使用旧的<em
浏览 10提问于2020-11-24得票数 0
3回答
用于消息处理的多个节点
、、、、
我们在Kubernetes上部署了一个spring引导应用程序来处理消息:它从Kafka主题中读取，然后执行一些映射，最后，它写到Kafka主题。为了获得更高的性能，我们需要更快地处理消息，因此我们引入了这个春季引导应用程序的多个节点。
消息包含一个状态。是否有任何解决方案来保持消息的有序性，并保证一个节点已经处理的<e
浏览 6提问于2020-04-24得票数 0
3回答
单个使用者订阅特定的主题，假设“Top-01”和"Group“为”Group-1“。Kafka与消费者的交互方式与Pub-Sub消息相同，直到新的消费者订阅了与“Group-1”相同的主题“Top-01”和“Group ID”。一旦新的消费者到来，卡夫卡将其操作方式转换为共享模式，并在两个消费者之间共享数据。这种共享将继续进行，直到使用者的数量达到为该特定主题配置的分区数为止。一旦用户<e
浏览 9提问于2017-06-22得票数 1
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
