Kafka :如何重新消费未提交/未确认的消息
Kafka是一个分布式流处理平台，它具有高吞吐量、可扩展性和容错性的特点。它主要用于处理实时数据流，支持发布-订阅模式，可以实现高效的消息传递和处理。
要重新消费未提交/未确认的消息，可以按照以下步骤进行操作：
- 确定消费者组的偏移量：Kafka使用偏移量（offset）来跟踪消费者在主题中的位置。首先，需要确定消费者组在主题中的偏移量，可以通过Kafka提供的工具或API来查询。
- 重置消费者组的偏移量：如果需要重新消费未提交/未确认的消息，可以将消费者组的偏移量重置为较早的位置。可以使用Kafka提供的工具或API来进行偏移量的重置操作。
- 重新启动消费者：在重置偏移量后，重新启动消费者，它将从指定的偏移量开始消费消息。消费者将会重新消费未提交/未确认的消息。
需要注意的是，重新消费未提交/未确认的消息可能会导致消息重复消费的问题，因此在实际操作中需要谨慎处理。可以通过在消费者端进行消息去重的方式来避免重复消费。
推荐的腾讯云相关产品是腾讯云消息队列 CMQ，它是一种高可靠、高可用的消息队列服务，可以满足分布式系统中的消息通信需求。CMQ提供了消息的持久化存储、高吞吐量、低延迟等特性，适用于各种场景，包括实时数据处理、日志处理、异步任务处理等。
腾讯云消息队列 CMQ产品介绍链接地址：https://cloud.tencent.com/product/cmq
相关·内容
1回答
我们使用手动提交策略。我们已经配置了一个bean，用于提交所有成功使用的消息。我们正在尝试添加一个逻辑来重新使用未确认的消息。我们面临的问题是，相同的消息被消费，但再次消费的时间间隔不统一。有时很快就会被重新消费，但有时需要几个小时。 我们如何确保未
浏览 48提问于2019-02-18得票数 0
我有一个业务场景，当消费者或kafka重新启动时，消费者不应该消费来自topic的已提交/未提交的消息。我尝试应用auto.offset.reset: latest.But其从topic.For中提取未提交的偏移量。例如，有一个应用程序具有一个具有1个主题和1个partition.Suppose的应用程序。我发布了10
浏览 17提问于2020-04-24得票数 0
我有一个Kafka使用者，其中使用HTTP POST调用将消息传递给另一个应用程序。我还使用以下命令手动提交偏移量 acknowledgment.acknowledge()； 有一些HTTP返回错误代码，我们忽略错误并提交偏移量，还有一些错误代码，我们没有提交偏移量。问题是，只有当我重新启动消费者时，kafka消费者才会轮询未提交的消息。如果分区中有未</
浏览 31提问于2021-06-26得票数 0
回答已采纳
我知道Kafka Queue Partition中的所有消息(或偏移量)都有它的偏移量，它负责偏移量的顺序。但是，如果我有一个Kafka Consumer Group (或单个Kafka Consumer)，它正在读取特别是Kafka主题分区，那么它如何维护读取的偏移量消息，以及谁维护这个偏移量计数器？如果消费者宕机，那么新的消费者将如何
浏览 18提问于2020-07-04得票数 1
回答已采纳
2回答
我们想知道对于给定的Kafka主题，还没有被Kafka Consumer消费和确认的消息数量。
有没有办法在Java中获得给定主题(主题有10个分区)中未消费的Kafka消息的计数?？
浏览 3提问于2019-10-24得票数 3
当我手动确认消费者应用程序中的偏移量时，我希望消费者可以自动重新获得未确认消息，但我不能成功。这是我的配置文件： stream: binder: za-nodes: ****
浏览 1提问于2017-01-25得票数 0
我使用的是Kafka 0.8.1。我正在使用一个带有多个分区的主题。避免处理分区的新节点和旧节点之间的竞争的</em
浏览 2提问于2015-11-24得票数 1
1回答
factory.setConsumerFactory(consumerFactory()); } // do not acknoledge the message here in order to retry it later.}
在条件为"false“的情况下，我们继续到"else”部分，我的消费者什么时候会再次尝试读
浏览 5提问于2020-09-02得票数 1
我希望将Solace保证的消息传递与负载平衡的消费者一起使用。目前，如果我使用具有非独占访问类型的队列，并且有多个消费者，则每个消费者将获得一个消息循环，并且如果一个消费者死亡，其未确认的消息将被转发给另一个消费者。这是我需要的功能，但是，由于消息有效负载的大小不同，我可能会超出限制，因此需要
浏览 2提问于2018-02-24得票数 0
1回答
我正在尝试实现一个循环，我们将在循环中处理一批消息，成功处理后，我们将提交消息的偏移量，但是对于下面的场景，我无法找到任何有形的输入。Read: 1，2，3，4，5...10我将依次循环处理它们，如果消息处理成功，则只应提交偏移量。所以想象一下，1,2成功，3失败，4,5成功，7,8失败...如果我只提交偏移量1,2,4,5，那么未提交的偏移量会发生什么? 3,7,8...
有什么想法会再读一遍吗？
浏览 1提问于2021-05-27得票数 1
让我们想象一下kafka分区中的消息流。消费者是否有一种方法来提交一条消息(例如2)，以便如果该特定的使用者重新开始读取消息，那么它只会使用未提交的消息？
浏览 0提问于2018-08-28得票数 0
回答已采纳
所以基本上我看到的是，我正在使用@KafkaListener消费消息，我故意不想承认它们，所以消费者忽略它们，并在主题中侦听新的消息。因此，如果应用程序重新启动，使用者只会从最后提交的消息开始。因此，我希望避免这种情况，我希望消费者在不重新启动的情况下考虑未提交的消息。下面是我正在使用
浏览 3提问于2019-12-05得票数 1
设置卡夫卡使用者的autocommit.enable选项将导致被消耗的消息被提交，这意味着如果使用者崩溃，它将开始从最后一个提交位置读取偏移量。但是，如果我们重新启动kafka服务器，消费者是否会重新读取已经提交的偏移量，或者在这种情况下，此选项也能工作--在服务器重新启动后，只会消耗未读消息？
浏览 2提问于2016-09-22得票数 0
回答已采纳
1回答
在Kafka 0.11.0.0的最新版本中，Apache团队引入了幂等生产者和事务。有没有可能保证我们想要记录的整个消息集(例如一百万条)只在结束时提交？我希望，例如，如果生产者失去了与经纪人的联系，并且不能重新稳定它，消费者将看不到任何消息。有可能吗？
浏览 16提问于2017-07-13得票数 1
回答已采纳
我的用例是使用Kafka使用者-api，这样我们就可以从一个kafka主题的最后一个成功处理的数据的偏移量中手动读取数据，然后手动确认是否对Kafka进行了成功处理的数据。(这是为了减少数据损失)。我是卡夫卡的新手，用下面的方式实现了我的消费者(我们使用的是弹簧引导)
的问题是:，即使我注释掉了ack.acknowledge()，偏移量仍在更新，
浏览 2提问于2020-01-21得票数 1
回答已采纳
1回答
我在scala中使用Alpakka-kafka来消费Kafka主题。下面是我的代码： ConsumerSettings(actorSystemkafkaConsumerSettings, Subscriptions.topics(kafkaConfig.topic))但是，消费者只从
浏览 5提问于2020-04-03得票数 1
高级消费者API似乎是一次读取一条消息。如果消费者想要处理这些消息并将其提交给其他下游消费者，比如Solr或Elastic-Search，这对他们来说可能是一个很大的问题，因为他们更喜欢批量发送消息，而不是一次只有一个。在内存中对这些消息进行批处理也不是一件容易的事，因为Kafka中的偏移量只有在批处理已经提交时才需要同步，否则具有未提交
浏览 2提问于2016-03-04得票数 9
1回答
我对Kafka非常陌生，我们使用的是Kafka 0.8.1。
我需要做的是从topic消费一条消息。为此，我必须用Java编写一个使用者，它将使用topic中的消息，然后将该消息保存到数据库中。保存消息后，将向Java使用者发送一些确认消息。如果确认为true，则应从主题中使用下一条消息。如果acknowldgement为false(这意味着由于
浏览 2提问于2014-07-31得票数 3
回答已采纳
我使用的是一个主题，一个分区，一个消费者，Kafka客户端版本是0.10。
但是，如果我首先生成消息，而没有提交它的偏移量，那么在几秒钟后暂停分区，以调用恢复方法。KafkaConsumer将不会接收未提交的
浏览 0提问于2018-11-12得票数 0
回答已采纳
我已经实现了一个简单的Kafka消费者，它实现了AcknowledgingMessageListener。在onMessage方法的和处有acknowledgment.acknowledge()；，当一切顺利或只有一个可恢复的异常时，它就会被调用。在这种情况下，一切工作正常。因此，应用程序崩溃，但在重启后，它没有收到来自受影响的Kafka主题的任何消息。即使Kafka重启也无济于事，应用服务器重启也无济于事等等，但是
浏览 2提问于2017-11-02得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
