Kafka消费者一次收到大量消息，我需要在提交第一条消息后才能陆续收到。我尝试了autoCommit:false，并尝试在接收数据时也使用pause()和resume()。但是，它仍然不起作用。
浏览 8提问于2018-08-04得票数 0
1回答
所以我从我的kafka用户那里得到了一些恼人的补偿提交案例。我在我的项目中使用'kafka-node‘。我创建了一个主题。在超过2台服务器的消费者组中创建了2个消费者。自动提交设置为false。对于我的消费者得到的每个消息，他们开始一个异步进程，这可能需要1到20秒，当进程完成时，消费者提交偏移量。我的问题是:有一个传感器，消费者1收到一条消息，需要20秒来处理。在处理过
浏览 2提问于2019-05-17得票数 0
我第一次在kafka中使用Node，使用的是kafka-node。消费消息需要调用外部API，这甚至可能需要一秒钟的响应时间。我希望克服我的消费者的突然故障，如果一个消费者发生故障，将取代它的另一个消费者将收到相同的消息，即其工作尚未完成。
我正在使用kafka 0.10，并尝试使用ConsumerGroup。我考虑在options中设置autoCommit: false，并仅在消息的工作完成后才提交消息</
浏览 0提问于2016-10-18得票数 4
我在NodeJS应用程序中使用Kafka-Node来产生和使用消息。我启动了一个等待我的主题的消费者。然后启动producer并将消息发送给Kafka。我的客户正在将这些消息中的每一条都插入Postgres数据库。当我停止消费者并继续生产时，我在大约30秒后重新启动消费者。我注意到大约有十几条消息已经从原始消费者插入到数据库中。
我假设当我杀死<
浏览 21提问于2019-07-11得票数 1
当生产者发送消息时，就会实时地使用消息.一切都如期而至。然而，最近在一台生产服务器上，Devops意外地将时区从PST更改为EST，作为OS修补程序的一部分。在Kafka服务器重新启动后，我们看到很少(不是全部，而是随机的)旧消息正在被消费者消耗。我们要求Devops将其更改为PST并重新启动。同样，这些旧信息也在本周末再次出现。此后，我们的消费者开始看到来自偏移量0的消息。调试后，我发现时区发生了变化，管理员在4天后从EST更改为P
浏览 0提问于2018-06-24得票数 3
我有下面的卡夫卡消费者代码，其中3个线程读取卡夫卡主题有3个分区。
有没有办法，只有在线程当前正在处理的消息得到处理后，才会从kafka主题中读取新的消息。例如，假设topic中有100条消息，那么是否可以一次只读取3条消息并对其进行处理。现在，当这3条消息被处理时，只需要读取接下来的3条消息，依此类推。
浏览 11提问于2017-03-01得票数 2
我正在开发一个推送通知，它向客户端发送许多消息。消息被发布到主题中，子脚本从同一主题中读取消息。如果在从主题偏移读取消息后出现错误，则递增，即使我无法发送消息，我的订阅者也需要读取下一条消息并发送它。如何阅读带有致谢的消息？
浏览 3提问于2020-10-20得票数 1
回答已采纳
高级消费者API似乎是一次读取一条消息。如果消费者想要处理这些消息并将其提交给其他下游消费者，比如Solr或Elastic-Search，这对他们来说可能是一个很大的问题，因为他们更喜欢批量发送消息，而不是一次只有一个。在内存中对这些消息进行批处理也不是一件容易的事，因为Kafka中的偏移量只有在批处理已经提交时才需要同步，否则具有未提交下游消息的崩溃的kafka
浏览 2提问于2016-03-04得票数 9
我有一个java Kafka消费者，在其中我正在批量获取ConsumerRecords进行处理。system which can take random time for different requests or timeout in 5 seconds.我遇到的问题是，如果产生了较晚的记录，但前一条记录仍未超时，则如何提交或提交哪个偏移量。假设我在一批中获得了2条记录，第一条消息的外部调用仍在等待，而第二次调用已完成。如果我等待5秒等待外部响
浏览 2提问于2018-01-05得票数 2
我使用的消费者组只有一个消费者，只有一个经纪人( docker wurstmeister image )。在代码中决定是否提交偏移量-如果代码返回错误，则不提交消息。我需要确保系统不会丢失任何消息-即使这意味着永远重试相同的消息(现在;)。为了测试这一点，我创建了简单的处理程序，它不会在'error‘字符串作为消息发送到kafka的情况下提交偏移量。所有其他字符串都被提交。现在我们传递'error
浏览 1提问于2018-03-19得票数 2
我正在尝试使用NET CORE 2.1制作一个消费者Kafka，这个消费者应该读取一条消息来比较时间戳和提交与否，所以这个消费者可以停留在相同的消息上，直到这个验证为真。Console.WriteLine($"Error occured: {e.Error.Reason}"); }
因此，我的主题中有10条消息我希望只有当我提交</e
浏览 10提问于2019-07-23得票数 0
我已经关闭了自动提交，并且在读取后不再从消费者那里提交偏移。
检查的消费者延迟也保持不变，它确保偏移量不会被提交。但问题是，它再次使用下一条消息而不是相同的消息。只有在前一个偏移量已提交的情况下，我才能读取下一条消息。请帮我做这件事。
浏览 0提问于2018-10-12得票数 4
我创建了一个生产者和一个消费者，使用“Kafka节点”包发送和消费来自kafka主题的信息。生产者和消费者通过API调用。POST方法用于向主题发送消息，而GET方法用于在使用者处从主题获取消息。当我在向KAFKA发送消息之后调用消费API时，所有以前的消息都在consumer.on('message', function(message){})中得到了安慰。我只需要最后一条信息，那是制作人发来的。producer
浏览 6提问于2021-12-15得票数 0
1回答
Kafka消息传递至少使用一次消息传递来确保要处理的每一条消息，并使用消息偏移来指示接下来要传递的消息。当有多个使用者时，如果某个致命的消息在消息处理过程中造成了使用者崩溃，该消息会被重新传递到其他使用者并传播死亡吗？如果某些慢速消息阻塞了单个使用者，其他使用者还能继续处理后续消息吗？或者更糟糕的是，如果一个慢而致命的消息导致了消费者崩溃，它会导致其他消费
浏览 5提问于2017-09-22得票数 0
回答已采纳
我有一个正在运行的Kafka集群，当重新启动一个应用程序(消费者)时，它会跳过一些在应用程序关闭时推送到主题的消息。当应用程序启动时，我可以看到它读取带有偏移量100的消息，然后将偏移量101推送到__consumer_offsets。然后，当应用程序关闭时，带有偏移量101, 102 and 103的消息将被推送到主题。在应用程序重新启动后，它读取101并将其偏移量设置为104，从而跳过102 and 103。
浏览 0提问于2020-04-21得票数 0
我计划使用带有固定线程池和IBM消息传递的ExecutorService编写类似于生产者和使用者的代码。
每个消费者工作者线程都接受这10条消息，就像每条消息一个工作线程一样？在将此消息作为上面的第二
浏览 2提问于2013-08-13得票数 1
回答已采纳
我有一个主题，我必须从kafka服务器读取，所以我只需要创建消费者，可以从kafka主题读取数据，我总是得到错误主题不存在。main.jsvar config = require('.The topic(s) [object Object] do not exist
at new TopicsNotExistError (C:\uilo
浏览 3提问于2016-06-21得票数 2
回答已采纳
首先，消息必须由集群中的每个节点使用，但只由一个节点生成。我知道我可以为每个节点分配一个组id，这可能是它的节点ID，这意味着每个节点都会收到消息。已经排序好了。但这里有一些问题。如果有关于这个主题的新消息，那么旧消息几乎是垃圾。如何才能将主题限制为当前最多只有一条消息？考虑到数据是节点启动所必需的，我需要消费关于该主题的最新消息，无论我以前是否消费过它，并且希望不会在每次启动服务器时创建唯一的组id。这是可能的吗?
浏览 0提问于2017-12-06得票数 0
我正在开发一个系统，其中要处理的消息在ActiveMQ中被推送。我有严格的要求，消费者必须按传入顺序处理消息。如果消费者的消息处理失败，则需要回滚/恢复，并无限次地重试。只有当一条消息处理成功时，消费者才需要提交并继续下一条消息。
如何防止回滚消息自动转发到DLQ，如何针对这种需求配置合适的重发策略？
浏览 11提问于2017-02-24得票数 2
考虑到SQS不是FIFO，而且我不确定我什么时候能在我的消费者端收到消息。此外，我可能会在同一数据上的前一条消息之前收到一条新消息，从而导致数据不一致在这样的同步场景中，SQS通常是一个好工具吗？
浏览 3提问于2016-09-26得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
