Kafka消费者无法阅读新味精？
我对Kafka相对来说是新的，我试图在主题上发送消息后产生消费者。
单个生产者在不同的分区上发送200个msg。
producer = KafkaProducer(bootstrap_servers=[f'{ip}:9092'],
value_serializer=lambda x:
dumps(x).encode('utf-8'))
for e in range(200):
data = {'count':e ,'time':time.time() }
producer.send('ner', value=data)
consumer = KafkaConsumer(
'ner',
bootstrap_servers=[f'{ip}:9092'],
auto_offset_reset='earliest',
enable_auto_commit=True,
group_id='ner_group',
value_deserializer=lambda x: loads(x.decode('utf-8'))
)
for message in consumer:
msg = message.value
time.sleep(1)
我多次运行消费者脚本。
共有1个答案
所有最初的200个味精将流向消费者-1，新的味精将流向消费者-2
这可能是你正在经历的，但这不是保证的行为。
如何在消费者组中添加新的消费者，以读取已经发送到某个主题的味精，而另一个消费者已经从该主题读取味精？
你所做的是正确的。添加另一个消费者不会读取已提交的组中的偏移量
-
我看到一个问题，我的主题中的所有消息都被我的消费者重新阅读。我只有1个消费者，我在开发/测试时打开/关闭它。我注意到，有时在几天没有运行消费者之后，当我再次打开它时，它会突然重新阅读我的所有消息。 客户端 ID 和组 ID 始终保持不变。我显式调用提交同步，因为我的启用.我确实设置了 auto.offset.reset=最早，但据我所知，只有在服务器上删除了偏移量时，才应该启动。我正在使用 IBM
-
因为我是新的Kafka，所以我能够从文件中读取记录，并通过生产者将消息发送到Kafka主题，但不能通过消费者消费相同的主题。 注意：您可以从任何文本文件中读取数据，我使用的是Kafka2.11-0.9。0.0版本 这是我的密码： 下面是输出：
-
我使用以下代码来读取主题的数据，即“sha-test2”，但它正在读取完全替代的代码行，即 10 行中的 20 行。但是当我运行控制台时，它显示所有 20 行。即.bin/kafka-console-consumer.sh --zookeeper 本地主机：2181 --主题 sha-test2 --从头 我做错了什么？非常感谢您的帮助。
-
我有一个简单的java制作人，如下所示 我正在尝试读取如下数据 但消费者并没有从Kafka那里读到任何信息。如果我在处添加以下内容 然后消费者开始从题目开始阅读。但是每次消费者重新启动时，它都从我不想要的主题开始读取消息。如果我在启动消费程序时添加了以下配置 然后，它从主题中读取消息，但是如果消费者在处理所有消息之前重新启动，那么它不会读取未处理的消息。 有人可以让我知道出了什么问题，我该如何解决
-
我的用例是使用kafka消费者api，这样我们就可以从kafka主题中手动读取最后一次成功处理的数据的偏移量，然后手动确认Kafka的成功处理数据。（这是为了减少数据丢失）。然而，在我当前的实现中，程序向前移动并从下一个偏移读取，即使我注释掉了“ack.acknowledge（）”。我是新来的Kafka和实现我的消费者下面的方式（我们使用Spring引导） 问题是：即使我注释掉ack.acknow
-
我们使用的是spring集成kafka版本3.1.2。RELEASE and int kafka：消息驱动的通道适配器，用于使用来自远程kafka主题的消息。生产者发送加密消息，我们使用反序列化器解密实际消息。我们可以使用主题中发布的所有消息。我们将自动提交用作false。我们想知道在成功处理消息后如何从我们的服务提交或确认消息。有人能帮助我们如何提交从消息驱动通道读取的消息并提供一些参考实现吗？