如果有发布者和N个使用者，如果消费者使用auto.offset.reset=latest，那么他们会错过在订阅主题之前发布到某个主题的所有消息.众所周知，使用auto.offset.reset=latest的消费者在订阅之前不会重播主题中存在的消息.所以我需要：
让发行者等到所有订阅者开始使用消息，然后开始发布。不知道如何在不利用动物园管理员的情况下做到这一点。卡夫卡提供了这样做的手段吗？另一种方法是让auto.offset.re
浏览 2提问于2016-06-14得票数 1
: EventHubClient(ADDRESS).subscribe(MyReceiver(str(partition)), CONSUMER_GROUP, str(partition), OFFSET
浏览 1提问于2018-04-16得票数 1
回答已采纳
我正在构建一个应用程序，我需要订阅和取消订阅卡夫卡主题的动态。问题是，我无法找到重新订阅主题的方法，因此只能获得订阅后出现的有关该主题的新消息。当设置"auto.offset.reset": "latest"时，我只会在第一次订阅时获得新消息，而对后一次订阅则不会收到新消息。
每次我需要订阅一个新的主题时，我是否应该创建一个新的消费群体？KAFKA_BOOTS
浏览 1提问于2021-05-04得票数 1
回答已采纳
假设我有一个主题"test“，它已经有了一堆消息，现在我想订阅这个主题。但我不想在新订阅之前使用所有发布的消息。基本上，我只想消费订阅后发布的最新消息。
浏览 12提问于2021-06-03得票数 0
我如何确认当前在谷歌PubSub精简版订阅积压中的所有消息。gcloud pubsub lite-subscriptions ack-up-to SUBSCRIPTION \
--location LOCATION --partition PARTITION --offsetOFFSET --project PROJECT 但我不知道如何设置相应的偏移量。
浏览 20提问于2021-11-05得票数 1
1回答
message #1234 of 5678 total in partition 0 of topic foo
我误解了RdKafka::Consumer->outq_len()和RdKafka::Topic->OFFSET_END
浏览 3提问于2015-08-28得票数 1
1回答
处理延迟的MQTT消息
、、、
发送设备通常是便携的，并且可能没有可靠的连接，因此消息很可能会延迟到达。
为了让程序正常运行，消息需要在一定程度上及时到达，比如20秒左右。如果消息晚于此时间到达，则需要进行一些特殊处理。我的问题是确定一条消息是否延迟，如果是的话，延迟多长时间。我想到的第一件事是发布者向消息添加一个时间戳，然后订阅者可以检查，但这取决于发布者和订阅者有一个同步的时钟。有没有其他方法可以同步发布者和订阅者之间的时间？另一种方法来确定我没有考虑到的消息有
浏览 0提问于2021-05-03得票数 0
1回答
使用Kafka，我们可以很容易地订阅(一个新的group.id)到一个主题，并开始消费事件。如果保留策略是默认的，那么如果指定auto.offset.reset=earliest，我们也可以使用一周前的消息。对吗？但是，如果我们想从开始，从开始呢？我想应该使用KTable，但我不知道当新客户机订阅了有状态流时会发生什么。你能告诉我新的订阅者会收到所有聚合的消息吗？
浏览 11提问于2022-09-08得票数 0
在python程序中，我想向Kafka写一些消息，然后读取来自不同主题的远程应用程序的相同数量的消息。问题是，当我完成发送消息时，另一端已经开始响应，当我开始阅读时，我只能获得消息批的尾部部分，或者根本没有消息，这取决于时间。这与我对包的理解相矛盾，即我认为如果我使用auto_offset_reset='latest'创建一个消费者，并订阅一个主题，那么它会在订阅时记住偏移量，当我迭代消费者对象时，它将开始从该偏移量读取消息
浏览 16提问于2019-09-16得票数 0
RocketMQ消费者可以订阅具有多个标签的主题，如下所示生产者如何同时使用taga和tagb发送消息Topic: handy_test
Ta
浏览 4提问于2017-04-12得票数 1
1回答
Kafka流在控制中心不显示输出
、、、、
我能够创建一个主题，用kafka-node发布消息，在KSQL编辑器中订阅kafka-console-subscribe.sh和CREATE STREAM。我还在查询编辑器中将auto.offset.reset设置为Earliest。少了什么？
浏览 2提问于2020-03-18得票数 0
如果使用者对来自主题(offset = OFFSET_END)的消息是最新的，那么消费者如何执行它的阻塞poll()
消费者是否默认遵守酒吧/订阅的心态，在这种心态下，它会睡觉，等待来自经纪人的发布和广播
浏览 4提问于2019-10-24得票数 1
回答已采纳
当添加主题订阅时，我希望每小时运行一个批处理作业，以获取所有新消息并将其推送到另一个数据存储中。如果我在晚上7点15分添加订阅，当批处理作业在晚上8点运行时，我希望从晚上7点15分开始的所有消息都被批处理。因此，我想要的解决方案是在添加订阅的瞬间获取主题的当前偏移量。我已经介绍了简单的使用者，但我不想涉及这个基本用例的所有集群管理方面。我
浏览 0提问于2014-11-21得票数 7
StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG
浏览 2提问于2021-01-26得票数 0
1回答
我尝试以100ms的速度发送/发布，消息看起来像这个x.x，所以大约每隔100ms左右就会调用一次订阅。我的问题是，我认为它还不够快(即，如果当前订阅还没有完成，而另一个订阅正在被调用/消息被发布) {
xaxis.Pan(xaxis.Transform(-1 + xaxis
浏览 16提问于2019-08-01得票数 0
当每个使用者启动时，它希望读取主题子集(称为历史主题)的最新历史消息，然后读取所有新消息。历史数据的确切起点并不重要，因为主要是获取很少编写的主题的信息。需要历史数据的主题将只有一个分区。我不想在寻找之前阅读任何消息，因为消息可能比我想要的要新。但是，在Kafka分配主题分区之前，至少必须调用Consumer.poll。 print(f"{partition.topic} has of
浏览 12提问于2022-08-23得票数 1
回答已采纳
在消费者中，当订阅一个主题并开始消费消息时，它从一开始就读取消息，有没有办法只读取未读的消息？这是我用来消费消息的代码。consumer.poll(100); System.out.printf("offset= %d, key = %s, value= %s\n", record.offset(), r
浏览 5提问于2017-06-08得票数 2
回答已采纳
1回答
let i = 0; let chunk = file.slice( offset, offset + chunkSize ); formData.append("fileUpload", chunk, file.name
浏览 20提问于2021-10-26得票数 0
回答已采纳
根据我的理解，以下代码应该订阅我的主题设置，偏移到开头： consumer.Assign(new TopicPartitionOffset(topic, Partition.Any, Of
浏览 0提问于2019-03-20得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
