2回答
我们正在努力并行我们的Kafka消费者，以处理更多数量的记录来处理峰值负载。我们已经在做的一种方式是通过在同一消费者组中旋转尽可能多的消费者和分区。我们的消费者负责进行API调用，到目前为止，它是同步的。我们觉得让这个API调用异步会让我们的消费者处理更多的负载。因此，我们尝试使API调用异步，并在其响应中增加偏移量。然而，我们看到了一个问题：
通过将API调用设置为异步，我们可以首先获得最后一条记录的响应，并且到那时前一条<
浏览 2提问于2019-06-19得票数 2
此外，我们还可以标记从分区读取信息的最后一点。这样，我们就有了一个检查点，如果我们必须返回并找到丢失的数据，就可以向前移动，而不必重新读取先前的数据。因此，如果我们设置每个事件的检查点水印，我们就不会丢失一条消息。
浏览 1提问于2018-07-30得票数 1
我正在编写一个Kafka消费者应用程序，其中我有一个消费者为每个分区。log.error("Commit failed for offsets {}", offsets, e); }}); 是否有一种以编程方式访问和打印使用者滞后偏移量的方法，或者以其他方式表示使用者读取的最后一条记录的偏移量与某个生产者写入该使用者分区的最后一条记录的偏移量之间的位置差异
浏览 1提问于2020-10-26得票数 1
回答已采纳
当我们向队列发布消息时，我知道将由谁来处理:生产者和消费者在一对一的关系中耦合(到目前为止这是正常的)。我现在想做更复杂的事情(消费者和生产者之间的关系)：
当我阅读
浏览 6提问于2014-05-21得票数 0
1回答
我想使用卡夫卡来执行有界的批处理，程序将知道它何时处理最后一条记录。批处理：
如果这是最后一次记录，则将批处理作业标记为在数据库中完成的作业。
浏览 0提问于2018-05-08得票数 0
当我从主题中收到一条消息时，它实际上是原始消息中每一行对应的一条消息，并且我不能将该消息转换为ConsumerRecord<String、String>。生成记录的代码如下所示：Kafka配置如下所示(这仍处于使用Testcontainers的集成测试阶段，因此生产者正在对消费者正在监听的同一主题进行生产)：最后，消费者逻辑：
<cod
浏览 64提问于2021-06-16得票数 0
回答已采纳
我在一个系统上工作，在这个系统中，客户端将数据输入到程序中，然后save操作将一条消息发送到activemq，以便进行更多的时间密集型处理。我们很少遇到这样的情况:客户端连续两次更新一条记录，而activemq队列中的消费者将同时处理这两条记录。我正在寻找一种方法来确保包含具有相同身份的记录的消息按顺序处理，并且一次只处理一条。为了清楚起见，如果ID为1、1和2(按此顺序)的记录被发送到activemq，则1将进行处理，然后是2(如果1仍在处理中)，<
浏览 6提问于2016-07-11得票数 0
1回答
我正在使用sparkstreaming从Kafka消费，想想如果在消费者端发生了一些崩溃或类似的事情。我不担心复制记录，但我不能错过任何一条记录。如果我重新启动我的消费者，我将不会从我离开的地方继续。但在最坏的情况下，我想检查一下从kafka主题中最古老的记录运行我的消费者的可能性。
浏览 2提问于2018-03-02得票数 0
我们在Activemq中有10条消息，我们只启动了2条consumers.But，用户首先消费和处理这些消息。第二，消费者不消费信息。如果我在第一次消费者处理时间的同时向Queue再发送一条消息，第二次消费者使用和处理该特定消息(我们在第一次消费者处理时间时发送了一条消息) only.After，那么它不会消耗挂起的混乱。最后，我所理解的是，所有挂起的消息都是由第一个消费者处理的，而不是剩余的消费者
浏览 3提问于2013-09-13得票数 0
回答已采纳
在我的一个测试中，我的应用程序使用KafkaTemplate和大多数默认配置设置将100条记录连续发送到EmbeddedKafka实例上的单个TopicPartion。我使用KafkaTestUtils.getRecords(消费者)方法尝试从Kafka实例获取记录，并验证它们是否都已发送。为什么getRecords
浏览 3提问于2016-08-12得票数 2
我已经为它创建了主题和消费者组。我在启用自动提交的情况下启动了consumer。因为我的消费者现在很慢，所以有很多延迟，并且由于它的自动提交，偏移量被存储起来。因此，我想从最后一条消息开始，但kafka是根据存储的最后一个偏移量进行消费的。我甚至用自动提交和自动偏移量存储禁用启动了消费者，但因为它以前是存储的。它总是从最后一个偏移量开始，而不是从最后一个消息开始。 如果我可以重置存储的偏移量，那么我计划在禁用自动提交的情况下使用，它应
浏览 113提问于2021-07-20得票数 0
我的消费者配置有kafka批量监听器配置和@KafkaListener消费消息列表。我有一个ConsumerInterceptor，我想为每条记录设置惟一的id，并将它的值存储在MDC中。如果我的kafka监听器消费了一条消息，那么唯一的id是正确的。但是我的kafka侦听器使用消息列表，所以MDC.get("id")只获取最后一个值。怎么处理呢？
浏览 14提问于2021-06-11得票数 0
因为我们的应用程序消耗的是公司范围内的主题，在这些主题中，只有一小部分事件对我们的应用程序很重要，因此消费者的延迟是巨大的--数十万用于一些特定的主题。我们检测消费者是否落后于事件消费的主要度量是消费者滞后度量(即检测我们存在吞吐量问题)。但是，没有提交跳过的消息的偏移量对此没有帮助。问题
我们想知道你们是如何发现消费者落后于消费(例如，消费者太慢)，以扩大消费者的数量？有没有办法为跳过的消息提交抵消？
浏览 2提问于2022-07-26得票数 1
我需要一些方法来获取订单ID的所有消息，以便发送给单个消费者。我想要避免的是，必须建立一个新的队列发送给每个消费者。我的消费者应该是有弹性的(随着负荷的增加，数量也在增加)。是否有一种方法可以让我的所有订单Is转到同一个消费者？(除了为每个消费者创建一个新队列之外)。
浏览 0提问于2019-09-04得票数 0
当我们准备好时，我们希望消费者将使用偏移量。由于文档的原因，我并不能真正理解seekToEnd代表什么。这是否意味着，它将查找到偏移量(最后一条ACKed消息)，还是将查找到主题的最后一条消息(无论偏移量如何)并等待新的消息？
浏览 0提问于2020-02-12得票数 0
假设我有一个来自join select的表，如下所示：TC0001 Book 15TC0001 Eraser 1TC0002 Eraser 1.5TC0003 Book 20
我想比较相同商品名称的上一
浏览 3提问于2015-09-01得票数 2
我轮询来自多个分区的记录，并且只提交一个记录，然后其余的记录似乎也被提交了，因为我不能再次轮询它们。并且我已经将auto.commit设置为false，但找不到原因。但是，如果该主题只有一个分区，它将正常工作，我在第一次轮询时没有提交的记录将再次重新轮询。当有多个分区时，为什么结果会不同？}
}
根据@mazaneicha的评论，我重新运行了程序，结果似乎我已经提交了所有记录，因为我没有跳出outter循环，但我的日志
浏览 15提问于2019-08-08得票数 1
是否可以将主题配置为只存储最后一条消息的副本并将其发送到新的连接，而不知道客户端标识符或其他信息？
停止消费
浏览 4提问于2014-02-26得票数 5
但是，我想知道如果队列中有多个消息，接收方如何才能检索最后一条消息？有没有我应该使用的特殊队列？谢谢,
浏览 4提问于2017-08-02得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
