我刚刚意识到，在Kafka主题上收到的消息丢失了标题。有没有办法在发布到Kafka Topic之前给消息添加头部，然后在消费者端阅读？我使用的是Java11，Spring Cloud Hoxton.SR6，Spring2.2.4，kafka_2.13-2.6.0。提前感谢！
浏览 0提问于2020-09-08得票数 0
如何限制kafka消费者的事件消费速度，使服务不受影响，.Getting kafka Topic中的海量数据，我需要处理所有的事件。我们有8个消费者从12个分区中读取了kafka主题中的大量数据，但保留时间为5天。那么，我如何从消费者方面进行限制，使消费者不会走下坡路。有没有办法在卡夫卡把消息的阅读</
浏览 105提问于2020-11-12得票数 0
仅出于测试目的，我想自动化需要检查Kafka消息内容的场景，所以只想知道是否可以使用Kafka java库直接从TOPIC读取消息，而不需要消费者？ 我刚接触Kafka，所以任何建议都对我有好处。
浏览 33提问于2020-07-24得票数 0
1回答
有时，kafka-node消费者从偏移量0开始消费，而它的默认行为是只消费较新的消息。则它将不会切换回其默认行为。你知道如何解决这个问题吗?知道发生了什么，它的行为突然改变了吗？var kafka = require("kafka-node") ; client = new kafka.KafkaClient();consumer =
浏览 5提问于2020-01-31得票数 3
在Kafka主题的消费者中，我很难找到处理异常的简单模式。场景如下:在使用者中，我调用一个外部服务。如果服务不可用，我想重试几次，然后停止消费。最简单的模式似乎是一种阻塞同步的方式来处理它，在java中是这样的：forcontinue;
} else thr
浏览 0提问于2018-04-28得票数 2
我知道我创建了测试主题，但我看到了一个名为"__consumer_offsets“的附加主题。从名称上看，它意味着它与消费者偏移量有关，但它是如何使用的呢？$ bin/kafka-topics.sh --list --zookeeper localhost:2181 __consumer_offsets test
$ bin/kafka-topics.sh:1 Configs:segment.bytes=104857600,cleanup.policy=compact
浏览 5提问于2018-12-13得票数 3
我正在使用segmentio/kafka-go客户端来读取主题中的消息。我找不到..如何从上一条/新消息开始阅读。
每次启动代码时，它都会从该分区中的起始偏移量开始读取。
浏览 8提问于2019-11-24得票数 2
是否可以使用0.10.0版本的KafkaConsumer从0.9.0Broker读取消息？但我正在寻找相反的方法:新的消费者，旧的经纪人。我有个例外：
org.apache.kafka.common.protocol.types.SchemaException: Erro
浏览 14提问于2016-06-01得票数 0
回答已采纳
我对Kafka非常陌生。我有一个要求，我只需要阅读一些特定的消息。
为此，我将向Kafka推送一条消息，其中包含一些键和值。有没有办法让消费者订阅特定的密钥列表。因此，当消息推送到Kafka时，如果消息具有消费者正在侦听的密钥，则消费者将进行消费。
浏览 2提问于2016-09-29得票数 2
1回答
有一个基本的例子，它对1个消费者来说就像一个护身符。它接收消息。但是添加一个额外的消费者将被忽略。let kafka = require('kafka-node');
let consumer1 = new kafka.Consumer(cl
浏览 4提问于2017-02-02得票数 2
回答已采纳
我的目标是从非文件来源(即在程序中生成或通过API发送)获取数据，并将其发送到spark流。1 --partitions 1 --topic my-topicPython 3.6.1| Anaconda custom (64-bit)我的问题是，当从使用者shell脚本检查主题时，没有显示任何内容：
$ bin/kafka-console-consumer.sh --bootstrap-s
浏览 14提问于2017-07-09得票数 0
回答已采纳
我们使用了3个群id相同的kafka消费者来读取数据。--replication-factor 5和--partitions 5的Kafka主题我们使用kafka版本= 0.10.0.1
这里的任何帮助都将不胜感激。
浏览 0提问于2018-03-26得票数 0
在我们的项目中，我们使用spring-cloud-stream-binder-kafka版本1.1.2与kafka集成。最近，我们遇到了这样一种情况:我们的一个服务在启动后使用了某个主题中的旧消息(已经使用过)。该主题有2个分区和2个消费者分组在一个消费者组下。我们不确定偏移量是否正确提交给zookeeper。在启动期间，每条消息都会抛出下面的错误消息</
浏览 14提问于2017-06-28得票数 0
我正在尝试编写一个简单的Kafka消费者来读取windows机器上的数据。但是我的消费者不能读取任何数据。生产者产生了20多条消息，但没有一条消息被消费。下面是我为消费者编写的代码： properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG通
浏览 0提问于2020-05-13得票数 1
2回答
假设分区有4个副本(1个领导者，3个追随者)，并且所有副本当前都是同步的。min.insync.replicas设置为3，request.required.acks设置为all或-1。生产者向领导者发送一条消息，领导者将其附加到日志中。在此之后，两个副本在获取此消息之前崩溃。剩下的一个副本成功获取消息并附加到它自己的日志中。领导者在特定的超时后会向生产者发送一个错误(我认为是NotEnoughReplicas)，因为不满足min.in
浏览 2提问于2016-07-01得票数 7
val actorSystem = ActorSystem("ReactiveKafka") bootstrapServers = "localhost:9092", groupId = "groupName",
valueDeserializer= new StringDeserializer()
浏览 7提问于2016-08-28得票数 0
回答已采纳
1回答
我想知道有没有Kafka Spout支持安全的Kafka Broker。来自apache storm的KafkaSpout不支持SSL。public KafkaSpout(SpoutConfig spoutConf) {}
浏览 3提问于2017-03-30得票数 1
我创建了一个生产者和一个消费者，使用“Kafka节点”包发送和消费来自kafka主题的信息。生产者和消费者通过API调用。POST方法用于向主题发送消息，而GET方法用于在使用者处从主题获取消息。当我在向KAFKA发送消息之后调用消费API时，所有以前的消息都在consumer.on('message', function(message){}
浏览 6提问于2021-12-15得票数 0
我的Kafka消费者的代码如下所示 topic = 'my-topic'
messages = consumer.poll(timeout_ms=kafka_config.poll_timeout_ms/
浏览 1提问于2018-04-19得票数 1
来自我的kafka生产者的一些消息超过了默认的1MB，并且被代理拒绝。
我无法访问代理和消费者来进行更改。因此，我只需要在生产者级别进行更改。我正在考虑在我的producer中使用batch.size到5MB的snappy压缩。请让我知道这是否是正确的选择或任何其他方式可用于我的用例。谢谢
浏览 0提问于2020-07-23得票数 0
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
