我正在使用Kafka python版本2.0.2来产生和消费消息:我的生产者： from kafka import KafkaProducerproducer.send('my topic', b'hello word') 我的消费</e
浏览 84提问于2020-12-03得票数 0
例如，我的kafka消费者(在consumer Group 1中)正在阅读来自Kafka主题A的消息。
现在，如果该消费者在失败之前消费了12条消息。当消费者重新启动，并且现在它有不同的消费者组(即消费者组2)时，问题1 -？在重启时，它会从偏移量(或位置)中停止的地方<em
浏览 1提问于2021-02-04得票数 0
我使用的是Kafka 0.8.1和Kafka python-0.9.0。在我的设置中，我有2个kafka代理设置。当我运行我的kafka消费者时，我可以看到它从队列中检索消息，并跟踪两个代理的偏移量。我的问题是，当我重新启动消费者时，它会从头开始使用
浏览 3提问于2014-07-10得票数 8
我正在使用segmentio/kafka-go客户端来读取主题中的消息。我找不到..如何从上一条/新消息开始阅读。
每次启动代码时，它都会从该分区中的起始偏移量开始读取。
浏览 8提问于2019-11-24得票数 2
1回答
我正在寻找一种将日志数据从文件流到我们的数据库的方法。我一直在阅读关于和管理实时的文章，但是我不知道如何管理非处理的数据，因为停止了。我的意思是，假设系统正在运行，数据是实时处理的，系统在10分钟后突然停止并重新启动。有没有一种方法可以在不影响实时流的情况下处理这些挂起的数据？
谢谢
浏览 2提问于2015-09-18得票数 1
回答已采纳
1回答
我在一个Kafka集群中消费一个Kafka主题(保留21天)，作为消费者(最早/从开始)连续15天与x消费者组在一起，在第15天，生产者团队停止生产，我在我这边停止消费者，以确保没有剩余的消息可供消费然后Kafka集群也被关闭了。然后，在第16天，Kafka Cluster启动了，P
浏览 29提问于2020-07-24得票数 0
我使用的消费者组只有一个消费者，只有一个经纪人( docker wurstmeister image )。在代码中决定是否提交偏移量-如果代码返回错误，则不提交消息。我需要确保系统不会丢失任何消息-即使这意味着永远重试相同的消息(现在;)。为了测试这一点，我创建了简单的处理程序，它不会在'error‘字符串作为消息发送到kafka的情况
浏览 1提问于2018-03-19得票数 2
test中总共有16条消息。--根据公认的答案，在消费群体group1中有一个消费者，该用户已经消耗了5条消息并死亡。下次启动这个使用者时，它甚至不会使用该auto.offset.reset配置，而是将继续从它死的地方继续，因为它将从偏移存储中获取存储的偏移量。根据Python文档- ，默认情况下，enable_auto_commit是True，这意味着用户的偏移
浏览 6提问于2017-10-05得票数 1
回答已采纳
我正在从基于JMS的系统转移到Kafka，Kafka负责在新系统和旧系统之间同步事务。新系统发布的每条消息都必须由订阅者/消费者成功处理。我不必担心消息的顺序。由于遗留系统中的一些设计问题(悲观锁定)，偶尔会有一些事务在消息到达时失败，在这种情况下，我希望消息在一段时间后返回。我正在试图弄清楚如何处理
浏览 2提问于2017-05-06得票数 0
我创造了一个项目来帮助我理解卡夫卡。它被设置为三个完全相同的Rails应用程序，它们都在Docker内部，而Karafka配置为使用这些消息--如果您在其中一个中创建了一个记录，它将被复制到另外两个。我认为start_from_beginning设置意味着每次重新启动Karafka服务器时，它都会从偏移量0开始，但情况似乎确实如此。有人能解释一下我做错了什么或者纠正我的理解吗？以下是karafka.rb的两个重要部分
setup
浏览 1提问于2019-12-31得票数 0
回答已采纳
Problem desc
无论多少个kafka节点，如果停止头两个kafka过程，消费者就无法从kafka获得消息(生产者可以向kafka发送消息)，经过检查，头两个kafka节点是消费者组协调器，如果一个挂起，另一个继续工作，如果挂两个，消费者都不工作，报告错误:偏移提交失败，这不是正确的协调者。3.我的问题
浏览 0提问于2018-06-07得票数 0
回答已采纳
我在服务器启动时启动Kafka consumer。如果Kafka集群长时间(几个小时)宕机，对消费者会有什么影响？Kafka再次上线后会收到消息吗？
浏览 46提问于2019-12-12得票数 0
我想使用Kafka高级消费者API，同时我想禁用偏移量的自动提交。我试图通过以下步骤来实现这一点。1) auto.commit.enable = false3) dual.commit.enabled = false1)高级消费<
浏览 1提问于2015-04-28得票数 5
2回答
我是Apache Nifi的新手，但我在尝试连接kafka微服务(与生产者)和Apache nifi消费者时遇到了一些问题。/test depends_on: command: python test.py 测试是我的制作人： from在Apache nifi仪表板中，我使用以下参数设置了一个输入: Kafka b
浏览 37提问于2021-07-14得票数 0
3回答
我在客户端使用kafka : kafka_2.12-2.1.0，spring kafka，遇到了一个问题。
我需要通过读取kafka主题中的所有现有消息来加载内存中的映射。为此，我启动了一个新的消费者(具有唯一的消费者组id，并将偏移量设置为earliest)。然后，我遍历消费者(poll方法)以获取所有<em
浏览 1提问于2019-02-11得票数 3
1回答
我对Kafka非常陌生，我们使用的是Kafka 0.8.1。
我需要做的是从topic消费一条消息。为此，我必须用Java编写一个使用者，它将使用topic中的消息，然后将该消息保存到数据库中。保存消息后，将向Java使用者发送一些确认消息。如果确认为true，则应从主题中使用下一条消息。如果acknowldg
浏览 2提问于2014-07-31得票数 3
回答已采纳
我在NodeJS应用程序中使用Kafka-Node来产生和使用消息。我启动了一个等待我的主题的消费者。然后启动producer并将消息发送给Kafka。我的客户正在将这些消息中的每一条都插入Postgres数据库。当我停止消费者并继续生产时
浏览 21提问于2019-07-11得票数 1
我发现，尽管kafka主题有很多消息(数百万)在排队，但vert.x消费者只获取了500条消息(默认的抓取量)，然后将这些消息传递给处理程序。但是，在消息被处理和提交之后，使用者只是停止并等待大约35秒，直到它获取另一批消息。 我预计使用者会继续获取，直到它设法赶上分区，然后暂停。我如何让它这样做呢？(result);
}); <
浏览 32提问于2021-03-26得票数 0
以下是我为kafka生产者编写的python代码，我不确定这些消息是否可以发布到Kafka代理。因为消费者端没有收到任何消息。当我使用生产者控制台命令对其进行测试时，我的Consumer python程序运行良好。RDDs.sentRDD.collect()
## Stop file connec
浏览 6提问于2016-09-13得票数 0
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
