当服务器重新启动后未使用消费者group.id时
，可能会导致以下问题：
- 消费者无法恢复之前的消费状态：在Kafka等消息队列系统中，消费者通过指定group.id来标识自己所属的消费者组。当服务器重新启动后，如果没有指定group.id，消费者将无法恢复之前的消费状态，无法继续消费之前未处理的消息。
- 消费者无法进行负载均衡：消费者组的概念是为了实现消息的负载均衡和高可用性。如果服务器重新启动后未使用消费者group.id，那么所有的消费者将被视为独立的消费者，无法进行负载均衡，可能导致某些消费者负载过重，而其他消费者处于空闲状态。
- 消息重复消费：在没有指定group.id的情况下，每个消费者都会从消息队列的起始位置开始消费消息。这可能导致消息被重复消费，特别是对于已经被处理过的消息。
为了解决以上问题，建议在服务器重新启动后使用消费者group.id来标识消费者组。group.id可以是一个唯一的字符串，用于标识一组具有相同消费逻辑的消费者。通过指定group.id，可以实现以下优势和应用场景：
- 消费状态的恢复：使用相同的group.id，消费者可以从上次消费的位置继续消费消息，确保消息不会被重复消费。
- 负载均衡和高可用性：多个消费者可以共享一个group.id，实现消息的负载均衡和高可用性。当某个消费者宕机或离线时，其他消费者可以接管其消费任务。
- 消费者的扩展和缩减：通过增加或减少消费者，可以根据实际需求动态调整消费者组的规模，实现弹性扩展和缩减。
对于腾讯云的相关产品和产品介绍链接地址，以下是一些推荐的选择：
- 腾讯云消息队列 CMQ：腾讯云消息队列 CMQ 是一种高可用、高可靠、高性能的分布式消息队列服务，可满足异步通信、流量削峰、解耦、日志处理、消息通知等场景需求。产品介绍链接：https://cloud.tencent.com/product/cmq
- 腾讯云云服务器 CVM：腾讯云云服务器 CVM 是一种可弹性伸缩的云服务器，提供高性能、可靠稳定的计算能力。可以根据实际需求选择不同配置的云服务器来部署应用程序。产品介绍链接：https://cloud.tencent.com/product/cvm
- 腾讯云云原生容器服务 TKE：腾讯云云原生容器服务 TKE 是一种高度可扩展的容器管理服务，支持自动化部署、弹性伸缩、服务发现和负载均衡等功能，适用于容器化应用的部署和管理。产品介绍链接：https://cloud.tencent.com/product/tke
请注意，以上推荐的产品仅供参考，具体选择应根据实际需求和项目情况进行评估和决策。
相关·内容
我在最新的文档页面中阅读了Spark Streaming kafka集成指南，它是基于Kafka 010版本的。
我可以看到其中一个卡夫卡参数是"group.id" -> "example"
我认为当我们使用DirectStream方法时，我们不需要将group.id作为一个参数传递。我对这个文档感到困惑。group.id和Spark Streaming Direct Stream的关系是什么？
浏览 0提问于2016-10-13得票数 1
1回答
我正在使用kafka (v0.10.1.1)和一些动态组(为每个新消费者创建新的group.id )。我必须重新检查/过滤过去的事件流，以获得一些标准，这意味着一段时间后，我有了许多组。有没有自动清理旧的未使用的组？
浏览 13提问于2017-02-21得票数 0
回答已采纳
当消费者实例group.id在重启时更改时，是否会对kafka性能产生任何影响。较旧的group.id将会发生什么情况?它是否仍然存在于代理内存中，或者何时将被删除？假设我有1000+消费者实例，并且所有实例都在重启时动态分配group.id。
可以为{log.retention.ms'}提供什么列表值。可以设置为1ms吗？
浏览 8提问于2018-08-28得票数 0
回答已采纳
我试图设置使用者组属性，使其只能每分钟使用一条消息。
consumer.properties
group.id=test_group
max.poll.interval.ms=60000
session.timeout.ms=60000
max.poll.records=1
生产者
kafka-console-producer.sh --broker-list localhost:6667 --security-protocol SASL_PLAINTEXT --topic cdc_poc
Consumer
kafka-console-consumer.sh --bootstrap-serv
浏览 2提问于2019-03-01得票数 1
回答已采纳
我正在按照快速入门指南上的说明在本地运行kafka。
然后，我在config/consumer.properties中定义了我的使用者组配置，以便我的使用者可以从定义的group.id中挑选消息
运行以下命令，
bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092
结果是，
test-consumer-group <-- group.id defined in conf/consumer.properties
console-consumer-67807 <-- when connecting
浏览 1提问于2020-05-13得票数 5
回答已采纳
在我们的应用程序中使用RabbitMq有成百上千的消息在时间上累积，对于几个队列处于未确认状态，TTLs不工作，因为当侦听器不存在时，有几次意外关闭。
现在，出于业务原因，我不能只是清空队列。我想手动将未确认的消息推送到就绪状态。
有吗?？
谢谢！
浏览 0提问于2016-12-02得票数 7
我在XD中有一个消费者作业，一旦它接收到由其他生产者job产生的消息，它就会完成。我每天都会触发这些工作。我发现这个消费者有时会收到一条以前用过的信息。
日志如下：
####OK
2019-06-28 02:06:13+0800 INFO inbound.job:Consumer_Job_In_XD-redis:queue-inbound-channel-adapter1 myConsumer.ConsumeTasklet - ==========consumed poll data ConsumerRecord(topic = my_consumer_topic, partition = 0
浏览 5提问于2019-07-11得票数 0
回答已采纳
这是我的消费者代码：
public class KafkaConsumer {
private ConsumerConnector consumerConnector = null;
private final String topic = "JsonTopic";
public void initialize() {
Properties props = new Properties();
props.put("zookeeper.connect", "localhost:2181&
浏览 0提问于2016-04-03得票数 0
当kafka消费者实例重启时，其消费者group.id在每次重启时都会发生变化。有这样的行为是好的吗，还是会影响kafka的性能？
我们可以为“log.retention.ms”属性提供的最小值是什么。
浏览 9提问于2018-08-27得票数 0
拥有一个在多个实例中运行的应用程序，并且每个实例都需要使用来自主题的所有分区的所有消息。
我知道有两种策略：
为每个应用程序实例创建一个独特的消费组id，并像往常一样订阅和提交，缺点是卡夫卡仍然需要代表每个用户维护一个消费者组。要求卡夫卡为这个主题提供所有分区，并为所有这些分区分配消费者。据我所知，卡夫卡不再有任何代表消费者的消费群体。因此，问题在于，由于卡夫卡方面没有消费者群体来跟上时代的发展，是否仍有必要做出补偿。创建消费者时没有指定“group.id p.id”.
浏览 0提问于2020-06-10得票数 0
我在我的Windows电脑上运行了和Apache服务器。问题在于Spring应用程序:每当我启动它时，它都会从Kafka读取相同的消息。这意味着偏移量没有被保存。我该怎么解决呢？
版本有：kafka_2.12-2.4.0，Spring Boot 2.5.0。
在卡夫卡的听众bean里，我有
factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
当Spring启动时，控制台上打印的使用者配置值如下：
2021-06-10 18:21:11.008 INFO 7036 --主o.a.k.clients.cons
浏览 1提问于2021-06-10得票数 1
回答已采纳
我在没有指定kafka组id的情况下运行pyspark structured streaming，每次运行程序时，我发现记录器消息中的group.id总是被更改。我搞不懂，如果它改变了group.id，它怎么知道下一次从哪里开始消费，或者我应该手动设置kafka的偏移量？我怎么知道它自动生成的主题id是否与我手动设置的其他主题id冲突？
浏览 2提问于2020-01-08得票数 0
2回答
有人知道如何在Python中使用相同的组id运行多个消费者吗？我试着追随
a = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'dd1',
'default.topic.config': {'auto.offset.reset': 'smallest'}})
b = Consumer({'bootstrap.servers': 'localhost:909
浏览 0提问于2017-10-22得票数 0
1回答
在启动confluent-dotnet使用者时，在调用订阅和随后的轮询之后，从服务器接收"Partition assigned“事件似乎需要很长时间，因此消息(大约10-15秒)。
起初我认为有一个自动主题创建开销，但无论消费者的主题/消费者组是否已经存在，时间都是一样的。
我使用此配置启动我的消费者，其余代码与融合的高级消费者示例中的代码相同：
var kafkaConfig = new Dictionary<string, object>
{
{"group.id", config.Cons
浏览 14提问于2018-01-12得票数 6
1回答
我想运行2火花结构的流式作业在同一个emr集群的消费者相同的卡夫卡主题。两个作业都处于运行状态。然而，只有一个工作可以得到卡夫卡的数据。我对kafka部件的配置如下所示。
.format("kafka")
.option("kafka.bootstrap.servers", "xxx")
.option("subscribe", "sametopic")
.option("kafka.security.protocol", "S
浏览 2提问于2020-08-01得票数 3
回答已采纳
我是Kafka的新手，我已经尝试过Kafka-Python包。
我设法设置了一个简单的生产者和消费者，它可以发送和接收消息。在这种情况下，消费者没有使用消费组，如下所示：
consumer = KafkaConsumer(queue_name, bootstrap_servers='kafka:9092')
但是，当我开始使用group_id时，如下所示，它停止接收任何消息：
consumer = KafkaConsumer(bootstrap_servers='kafka:9092', auto_offset_reset='earliest',
浏览 1提问于2019-03-19得票数 0
1回答
我可以针对Kafka location安装在命令行上发送和接收消息。我还可以通过Java代码发送消息。这些消息显示在Kafka命令提示符中。我还有一个用于Kafka消费者的Java代码。代码昨天收到消息。然而，它今天早上没有收到任何消息。代码未更改。我想知道属性配置是不是不太正确。下面是我的配置：
制片人：
bootstrap.servers - localhost:9092
group.id - test
key.serializer - StringSerializer.class.getName()
value.serializer - StringSerializer.class.ge
浏览 4提问于2016-06-24得票数 0
我在我的服务器上安装了kafka，并想学习如何使用它，我找到了一个由scala编写的示例代码，下面是其中的一部分。
def createConsumerConfig(zookeeper: String, groupId: String): ConsumerConfig = {
val props = new Properties()
props.put("zookeeper.connect", zookeeper)
props.put("group.id", groupId)
props.put("auto.offset.
浏览 6提问于2015-12-28得票数 6
3回答
我有一个Java Akka应用程序，从Kafka读取，处理消息，并手动提交。
我使用的是0.10.1.1 API的高级使用者。
奇怪的是，当我关闭应用程序并重新启动它时，偏移量比上次提交时要大一点，我找不到原因。
我在代码中只有一个提交点。
else if(message.getClass() == ProcessedBatches.class) {
try {
Logger.getRootLogger().info("[" + this.name + "/Reader] Commiting ...");
浏览 1提问于2017-05-25得票数 1
例如，我的kafka消费者(在consumer Group 1中)正在阅读来自Kafka主题A的消息。
现在，如果该消费者在失败之前消费了12条消息。
当消费者重新启动，并且现在它有不同的消费者组(即消费者组2)时，问题1 -？在重启时，它会从偏移量(或位置)中停止的地方继续，因为该偏移量是由Kafka和/或ZooKeeper存储的，还是会从第一条消息开始消费消息。
问题2->有没有办法确保在重启时(当消费者有不同的消费群体时)，它仍然从重启之前停止的地方开始消费？
为了给你提供上下文，我正在尝试在收到关于kafka主题的消息时更新每个节点/服务器中的内存缓存。为此，我对每个节点/服务器
浏览 1提问于2021-02-04得票数 0
1回答
我设置了一个节点kafka，并尝试了这样一个简单的pub/sub模式：
在我的笔记本电脑上，我通过代码生成了一些消息：
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.23.152:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384
浏览 4提问于2016-06-15得票数 3
1.问题描述
如果停止一个消费者并立即启动该消费者，则该消费者组需要大约10秒才能进行再平衡。
但是，如果停止一个消费者，等待大约10s，然后启动该消费者，这个问题将不会出现。
2.环境规划署( Env )
kakfa-版本:0.11.0.1
3.演示
Producer1Demo.java(生产者-向主题发送消息:主题1)
Producer2Demo.java(生产者-B发送消息到主题:主题2)
Consumer1Demo.java(消费者-从主题获取消息:topic1)
Consumer2Demo.java(Consumer从主题:topic2获取消息)
消费者-A和
浏览 0提问于2018-01-25得票数 3
回答已采纳
如何使用直播流API指定kafka spark流的消费组id。
HashMap<String, String> kafkaParams = new HashMap<String, String>();
kafkaParams.put("metadata.broker.list", brokers);
kafkaParams.put("auto.offset.reset", "largest");
kafkaParams.put("group.id", "app1");
JavaPa
浏览 0提问于2016-04-09得票数 7
回答已采纳
我有多个使用者，它们都使用相同的group.id侦听特定的主题。该主题有一个分区。
我的理解是，来自相同消费者组(由相同的group.id标识)的消费者将以循环方式获得消息，这样一条消息只由一个消费者处理。使用者在不同机器上的不同Windows服务中运行。
消费者是基于用C#编写的。
配置如下：
var config = new Dictionary<string, object>
{
{"group.id", "MyConsumerGroupId"},
{"en
浏览 1提问于2017-12-28得票数 0
1回答
我用scala编写了kafka consumer。当我运行consumer时，控制台上显示为空白。我使用了以下代码：
val topicProducer = "testOutput"
val props = new Properties()
props.put("bootstrap.servers","host:9092,host:9092")
props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeseria
浏览 0提问于2018-03-27得票数 0
我注意到Consumer配置有两个ID。一个是group.id (强制)，另一个是consumer.id (非强制)。
这两个ID有什么区别？
浏览 2提问于2016-01-01得票数 45
我正在使用Kafka站点上的ConsumerGroupExample代码测试Kafka High Level Consumer。我想检索Kafka服务器配置中名为"test“的主题的所有现有消息。看看其他博客，auto.offset.reset应该设置为“最小”，才能获得所有消息：
private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) {
Properties props = new Properties();
props.put("
浏览 1提问于2014-02-14得票数 7
我面对并发布了与KafkaConsumer相关的内容，我们的场景如下:我们有5个环境实现了KafkaConsumer，它们都指向相同的Kafka服务器和主题，所有的使用者都有相同的配置和group.id。
我注意到一些环境正在丢失消息，但是这些丢失的消息到达了其他环境。我认为这与我使用相同的group.id有某种关系。
例如，如果消息'A‘存在于env1中，则不存在于env1 2、3、4、5中。
谁能告诉我什么可能是原因，或者是否与group.id有关。
浏览 1提问于2022-05-20得票数 0
回答已采纳
我刚接触Kafka，我并不真正理解Kafka配置的含义，有没有人能给我解释一下更容易理解！
下面是我的代码：
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "master:9092,slave1:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
&#
浏览 0提问于2017-07-05得票数 21
我需要在发布/订阅模式下调用Kafka consumer 1000次。据我所知，为了让kafka在发布/订阅模式下工作，我需要给每个消费者一个新的groupId ( props.put("group.id"，kafka但是当我这样做时，如果两个消费者线程在同一毫秒内访问消费者，就会出现问题。这个问题该如何解决呢？
浏览 1提问于2018-05-13得票数 1
1回答
var configs = new Dictionary<string, string>
{
{"bootstrap.servers", MY_SERVER},
{"security.protocol", "SASL_PLAINTEXT"},
{"sasl.mechanism", "SCRAM-SHA-256"},
{"sasl.username", "MY_USERNAME"},
{"sasl.password",
浏览 4提问于2020-10-22得票数 2
回答已采纳
我已经使用以下属性设置了kafka consumer Properties consumerProperties = new Properties();
consumerProperties.put("bootstrap.servers",server);
consumerProperties.put("group.id",groupId);
consumerProperties.put("security.protocol", "SASL_PLAINTEXT");
consumerPro
浏览 42提问于2018-02-14得票数 2
回答已采纳
1回答
我有两个不同客户ID和组ID的消费者。除了保留时间和最大分区，我的Kafka安装包含默认配置。我环顾四周，看看是否有其他人有同样的问题，但不能取得任何结果。
所以情况是这样的：
消费者A:连接到Kafka，消耗大约300万条需要消耗的信息，然后闲置着等待更多的消息。
消费者B:不同的客户/组ID，连接到相同的Kafka主题，这导致消费者A获得300万条消息的重复，而消费者B也使用它们。
这两个使用者是两个完全不同的Java应用程序，不同的客户端和组ID在同一台计算机上运行。卡夫卡服务器在另一台电脑上。
这是卡夫卡的正常行为吗？我完全不知所措。
以下是我的使用者配置：
bootstrap.ser
浏览 2提问于2017-04-07得票数 0
回答已采纳
1回答
我正在创建一个使用python中的合流-kafka的使用者，我想以这样一种方式创建它:如果使用者被重新启动，它从主题中的最后一个可用消息(每个分区)开始，如果它在没有提交的情况下隐藏在消息后面并不重要。
这是为了避免处理数百万条消息，这些消息是在使用者关闭时生成的，并且不再需要处理。
我尝试设置参数auto.offset.reset的不同选项，但最多从上次提交的偏移量开始。这是我的配置：
consumer = Consumer({"bootstrap.servers": "localhost:9092",
"gr
浏览 3提问于2021-06-17得票数 0
6回答
了解消费群体id
、、
我重新安装了Apache 0.10.1.0。
我能够在命令提示符下发送/接收消息。
在使用Producer / Consumer示例时，我无法了解消费者示例上的group.id参数。
让我知道如何解决这个问题。
下面是我用过的消费者示例：
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
浏览 4提问于2016-12-29得票数 42
2回答
我使用的是来自的消费者。kafka版本为0.10.1.0。
下面是我的消费者的配置：
kafkaClient, err := kafka.NewConsumer(&kafka.ConfigMap{
"bootstrap.servers": broker,
"group.id": "udwg20",
"session.timeout.ms": 60000,
"go.events.
浏览 0提问于2018-10-02得票数 2
我已经写了一个非常简单的Flink流作业，从卡夫卡使用FlinkKafkaConsumer082的数据。
protected DataStream<String> getKafkaStream(StreamExecutionEnvironment env, String topic) {
Properties result = new Properties();
result.put("bootstrap.servers", getBrokerUrl());
result.put("zookeeper.connect", ge
浏览 0提问于2015-11-03得票数 2
我有多个产生消息的服务器，我需要一个服务器上的代理和使用者。如果我让生产者和消费者同时运行在同一台服务器上，那么它可以正常工作，但不确定需要做什么更改才能保持生产者隔离。我不想任何依赖动物园管理员和卡夫卡服务器在生产者服务器，因为有很多，他们会增加。在设置KafkaProducer时，我尝试将引导服务器更改为代理/使用者服务器，比如192.168.0.1:9092，但仍然无法生成消息。不知道我错过了什么，请帮帮我。我已经跟踪了的代码。
试着在同一台服务器上运行生产者和消费者，它运行得很好。
Producer.java
public class Producer {
public sta
浏览 0提问于2019-06-19得票数 0
我提交了多个spark streaming作业，这些作业使用相同的group.id消耗同一个kafka topic，正如纯Kafka文档中所说的，拥有相同group.id的多个消费者将加入同一个消费组，并且kafka topic中分块的记录将被拆分成这些消费者。然而，正如我在我的工作中测试的那样，这两个spark streaming作业仍然使用相同的主题的所有分区(不是均匀分割的)，并且在who进程期间没有发生重新分区/异常。这里有人知道Spark是如何管理与纯kafka平台不同的kafka分区偏移的吗？也许这是因为zookeeper管理spark中的kafka偏移量，而不是kafka本身管
浏览 63提问于2021-07-30得票数 0
1回答
我使用的是单节点Kafka broker (v0.10.2)硬件(8核，16 GB内存，1 TD HardDisk)和动物园管理员(v3.4.8)。我有一个有200个分区的主题，其中包含300万条消息。它花了5天的时间来完全处理所有的消息，一旦消息被处理，即Kafka-消费群。of在主题的所有分区中显示出0滞后，6小时后我停止了使用者.but，它显示了200万条消息的滞后，我发现它是重复的消息。这件事经常发生。我的补偿储存在卡夫卡经纪人身上。我的服务器配置是：
broker.id=1
delete.topic.enable=true
#listeners=PLAINTEXT://:9092
#
浏览 0提问于2017-05-19得票数 1
回答已采纳
我从日志中看到，665次使用了完全相同的消息。这一切为什么要发生？
我在日志里也看到了这个
Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.
This means that the time between subsequent calls to poll() was longer than the configured session.timeout.ms, which typically implies
that
浏览 2提问于2018-08-08得票数 5
回答已采纳
2回答
我已经启动了我的zookeeper和Kafka服务器。我启动了我的Kafka制作器，它发送了10条主题为'xxx‘的消息。然后叫住了我的卡夫卡制作人。现在我启动了我的Kafka用户，并订阅了主题'xxx‘。我的消费者使用我的Kafka生成器发送的10条消息，该生成器现在没有运行。我需要我的Kafka消费者应该只消费运行Kafka服务器的消息。有什么方法可以做到这一点吗？以下是我的消费者属性中的内容。
props.put("bootstrap.servers", "localhost:9092");
String consumeGroup
浏览 1提问于2016-09-30得票数 3
我们有主题，保留为7天(168小时)。当生产者发送消息时，就会实时地使用消息.一切都如期而至。然而，最近在一台生产服务器上，Devops意外地将时区从PST更改为EST，作为OS修补程序的一部分。
在Kafka服务器重新启动后，我们看到很少(不是全部，而是随机的)旧消息正在被消费者消耗。我们要求Devops将其更改为PST并重新启动。同样，这些旧信息也在本周末再次出现。
在较低的环境(开发、质量保证、阶段等)中，我们还没有看到这个问题。
卡夫卡版本: kafka_2.12-0.11.0.2
任何帮助都是非常感谢的。
添加更多信息..。最近，我们的CentOS进行了补丁更新，不知何故，管理员从P
浏览 0提问于2018-06-24得票数 3
我有一个包含5个分区的主题。我有5个线程，每个创建一个消费者所有的消费者都是相同的消费组使用group.id。我还为每个消费者提供了一个不同且唯一的client.id
我看到两个消费者正在阅读相同的消息来处理，kafka应该处理这个吗？如何对其进行故障排除？
浏览 36提问于2017-12-19得票数 1
我使用的是Confluent5.2的社区版本。我创建了一个名为employee-info的新主题。
我使用flink作为上述主题的消费者。在flink中，为了添加kafka源，我们需要传入group.id (消费者组)作为参数。如何为我的主题创建一个新的消费组？我很少有参考资料说在--consumer-property group.id=group_name中使用kafka-consumer-group。但是，当我运行kafka-consumer-group --help时，我看不到选项
浏览 45提问于2019-08-02得票数 1
Kafka服务器版本= 0.8.2.2
(Kafka c++ client)版本= 0.9.1
例如，一组中有两个消费者。两个消费者必须从主题中获得不同的消息。
我使用类似于下面，但似乎不起作用。两个消费者从主题中获取所有消息。
std::string topic_str = "sample";
std::string errstr;
int32_t partition = 0;
int64_t start_offset = RdKafka::Topic::OFFSET_END;
RdKafka::Conf *conf = RdKafka::Conf::cre
浏览 1提问于2016-10-05得票数 1
回答已采纳
1回答
我正在尝试编写一个Kafka消费者，以便从一开始就使用消息。我可以在控制台消费者中使用-- from - same来做同样的事情 但是我在JAVA API中找不到相应的属性。 def consumeFromKafka(topic: String) = {
val props = new Properties()
props.put("bootstrap.servers", "localhost:9092")
props.put("key.deserializer", "org.apache.kafka.comm
浏览 16提问于2020-04-06得票数 1
回答已采纳
我正在使用卡夫卡9，并混淆了订阅的行为。
为什么它期望group.id与订阅。
我们是否需要使用commitSync手动提交偏移量。即使不这样做，我也看到它总是从最新开始。
有没有办法从一开始就重播这些消息。
浏览 5提问于2017-02-02得票数 1
回答已采纳
我试图在每次调用消费者时重置消费者offset，这样当我多次调用消费者时，它仍然可以读取生产者发送的记录。我设置了props.put("auto.offset.reset","earliest");并调用了consumer.seekToBeginning(consumer.assignment());，但是当我第二次调用消费者时，它将不会收到任何记录。我该如何解决这个问题呢？
public ConsumerRecords<String, byte[]> consumer(){
Properties props = new Propert
浏览 0提问于2018-05-08得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
