kafka流从kafka主题中获取记录调用了多少次poll()
Kafka是一种分布式流式处理平台，它主要用于高吞吐量的、可持久化的消息传输，提供了一种高效、可扩展的方式来处理和存储大规模数据流。
针对你提出的问题，要计算Kafka流从Kafka主题中获取记录调用了多少次poll()方法，我们需要了解Kafka的工作原理。在Kafka中，消费者通过调用poll()方法从指定的Kafka主题中拉取消息。
每次调用poll()方法都会触发一次消费者向Kafka服务器发送请求，以获取一批消息。因此，我们可以通过统计调用poll()方法的次数来近似计算Kafka流从主题中获取记录的次数。
在实际情况下，调用poll()方法的次数取决于多个因素，包括消费者的配置、主题中消息的分区数量、分区中消息的数量等等。但是，由于我们无法获取到具体的运行环境和配置参数，无法准确给出调用poll()方法的具体次数。
然而，可以通过调整消费者的配置来优化poll()方法的调用次数，以提高消费性能和效率。例如，可以通过增加每次poll()方法调用返回的消息数量来减少调用次数，或者通过调整poll()方法的超时时间来控制调用的频率。
腾讯云提供了Kafka相关的云服务产品，可以满足各种不同规模和需求的用户。推荐的腾讯云产品是腾讯消息队列 CMQ，CMQ 提供了高可靠性、高可用性和高可扩展性的消息队列服务。更多关于腾讯消息队列 CMQ的信息可以查看腾讯云官网：腾讯消息队列 CMQ。
相关·内容
我正在尝试更多地理解kafka stream processor。我想知道kafka流处理器从kafka获取数据的frequency of polling是什么。 据我所知，kafka流处理器内部创建了一个kafka Consumer客户端，该客户端从kafka获取数据(并调用poll())。 那么当第一次调用poll()时，下一次会在什么时候再次调用它来从kafka获取数据呢？它每秒会发生很多次吗？我怎么知道已经调用了poll()多少次(为了从kafka获取数据)。 理解这一点的任何帮助都将是非常有帮助的。
浏览 43提问于2019-05-30得票数 3
回答已采纳
我的火花流版本是2.0，kafka版本是0.10.0.1，火花流-kafka-0-10_2.11。我使用直接的方式获得卡夫卡的记录，我现在想限制我在一批中获得的信息的最大数量。因此，我设置了max.poll.records值，但它不起作用。星星之火中的消费者数量是卡夫卡中的分区数?那么火花流中记录的最大数量是max.poll.records*consumers？
浏览 1提问于2018-09-27得票数 2
1回答
我在使用卡夫卡时遇到了一些问题。任何帮助都是非常感谢的！我有动物园管理员和卡夫卡集群3个节点在码头群。您可以看到下面的Kafka broker配置。
KAFKA_DEFAULT_REPLICATION_FACTOR: 3
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
KAFKA_MIN_INSYNC_REPLICAS: 2
KAFKA_NUM_PARTITIONS: 8
KAFKA_REPLICA_SOCKET_TIMEOUT_MS: 30000
KAFKA_REQUEST_TIMEOUT_MS: 30000
KAFKA_COM
浏览 0提问于2019-02-18得票数 1
1回答
我有一个消费者，它消耗了一个话题。主题有6个分区。分配给组的单个使用者。我做投票，如下面的Consumer.poll(10000)，我退出消费者获取时，没有记录返回。
从文档中，我相信当没有要使用的记录和持续时间10000足以重新平衡和获取记录时，轮询返回为空。多数时间轮询使用所有分区的记录，但有时轮询从3个分区获取记录，然后返回空记录，而不消耗其他3个分区。
顺便说一下，我使用了2.0.1Kafka客户端，Kafka服务器版本是2.11-2.2.0。
任何人都知道为什么我的使用者跳过其他分区并返回空的records.what来使用所有分区。
浏览 1提问于2020-01-23得票数 1
回答已采纳
2回答
我有一个火花流应用程序和卡夫卡流应用程序并排运行，为基准测试的目的。两者都使用相同的输入主题，并写入不同的目标数据库。输入主题有15个分区，火花流和kafka流都有15个消费者(1:1比例)。此外，事件有效载荷约为2kb。不确定它是否相关，但是90%的星火流执行时间是9ms左右。卡夫卡流，12毫秒。每次处理消息时，将在我的处理器中调用commit()方法。
这个问题依赖于高爆发。火花流可以保持700每秒，而卡夫卡流，大约60/70每秒。我不能超越这一点。见下图：(绿线-火花流/蓝线-卡夫卡流)
按照下面的配置，只要每个用户不超过1000个事件，考虑到背压，火花流就可以跟上，而不管每个
浏览 4提问于2020-06-23得票数 6
回答已采纳
如果生成的记录失败，我希望将回调设置为被触发。最初，我只想记录失败的记录。
Confluent Kafka python库提供了添加回调的机制：
produce(topic[, value][, key][, partition][, on_delivery][, timestamp])
...
on_delivery(err,msg) (func) – Delivery report callback to call (from poll() or flush()) on successful or failed delivery
如何才能在kafka-python 中实现类似的行为
浏览 0提问于2017-09-24得票数 6
回答已采纳
注意:使用kafka_2.11-0.9.0.1
我用3 Partitions创建了一个名为：consumer-tutorials的kafka主题，如下所示：
C:\kafka_2.11-0.9.0.1>.\bin\windows\kafka-topics.bat --describe --topic consumer-tutorial
--zookeeper localhost:2181
Topic:consumer-tutorial PartitionCount:3 ReplicationFactor:1 Configs:
Topic: consu
浏览 2提问于2016-06-21得票数 0
2回答
我已经看到了一个类似于的问题
但是，我仍然想知道来自特定分区的流数据是否不可能实现？我使用了卡夫卡消费策略在星火流订阅方法。
ConsumerStrategies.Subscribe
这是我尝试订阅主题和分区的代码片段，
val topics = Array("cdc-classic")
val topic="cdc-classic"
val partition=2;
val offsets=
Map(new TopicPartition(topic, partition) -> 2L)//I am not clear with this lin
浏览 0提问于2018-06-07得票数 0
回答已采纳
1回答
我有一个关于卡夫卡自动提交机制的问题。我使用Spring并启用了自动提交。作为一个实验，当系统空闲时，我断开了消费者与Kafka的连接30秒(主题中没有新消息，也没有消息正在处理)。在重新连接之后，我收到了如下几条消息：
Asynchronous auto-commit of offsets {cs-1915-2553221872080030-0=OffsetAndMetadata{offset=19, leaderEpoch=0, metadata=''}} failed: Commit cannot be completed since the group has alre
浏览 4提问于2020-09-24得票数 4
我使用汇合REST代理调用Kafka。我正在读取一个CSV文件，从所有记录中创建一个对象(大约400万条记录)，并向REST代理发送一个请求。我一直在获取一个OutOfMemory异常。
确切的例外消息是：
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "kafka-producer-network-thread | producer-81"
我只有一个REST代理服务器的实例，作为一个码头容器托管。环境变量设置为：
JAVA_OPTIONS=-X
浏览 0提问于2019-04-03得票数 2
回答已采纳
我有Kafka流应用程序与1.0.0 Kafka流API。我有单个broker 0.10.2.0 kafka和单个分区的单个主题。除生产者request.timeout.ms外，所有可配置参数均相同。我为producer request.timeout.ms配置了5分钟来修复问题。
在我的流应用程序中，我从Kafka读取事件，处理它们，然后转发到同一个kafka的另一个主题。
在计算完统计数据后，我观察到处理过程占用了5%的时间，其余95%的时间用于读写。
尽管我在Kafka上有数千万个事件，但有时Kafka民意调查返回个位数的记录，有时Kafka民意调查返回数千条记录。
有时上下文前进花费更
浏览 2提问于2017-12-19得票数 1
此接口- rd_kafka_poll_set_consumer()的用途是什么？在rdkafka.h文件中，它指出一旦您将rd_kafka_poll队列(rk->rep)重定向到rd_kafka_consumer_poll的队列(rkcg->rkcg_q)，在此之后您就不应该调用rd_kafka_poll。
谁能详细解释一下哪个队列会发生什么？producer是否应该调用rd_kafka_consumer_poll来查看生成的消息？
浏览 0提问于2016-10-01得票数 0
当使用者在5分钟内没有接收到消息(默认值为max.poll.interval.ms 300 000 of )时，使用者在没有退出程序的情况下就停止了。使用者进程挂起，不再消耗任何消息。
记录下列错误消息
MAXPOLL|rdkafka#consumer-1| [thrd:main]: Application maximum poll interval (300000ms) exceeded by 255ms (adjust max.poll.interval.ms for long-running message processing): leaving group
我看到，ErrMaxPol
浏览 0提问于2019-06-24得票数 14
1回答
使用spring cloud dataflow，我尝试使用Kafka压缩主题从数据库复制数据。出于这个原因，我需要在向主题(称为foo)中生成记录时设置记录的键。
kafka-topics --zookeeper localhost:2181 --create --topic foo --replication-factor=1 --partitions 3 --config cleanup.policy=compact
stream create --name foo --deploy --definition "jdbc --spring.datasource.url=jd
浏览 48提问于2018-08-03得票数 1
我正在使用Spring Kafka和错误处理(org.springframework.kafka.listener.ErrorHandler)，但我不清楚的是，当poll返回多条记录并且其中只有一条记录导致异常时会发生什么。据我所知，其他记录都被跳过了。如何实现准时的错误处理(例如，仅跳过受影响的记录并继续其他记录)？
浏览 7提问于2020-09-21得票数 0
回答已采纳
我使用了下面的Kafka消费者示例代码：
KafkaConsumer kconsumer = new KafkaConsumer(props);
ConsumerRecords<String, byte[]> records = kconsumer.poll(timeoutInMS);
kconsumer.commitSync();
我有两个问题：
(1)偏移量提交的地方是Kafka还是Zookeer？(我知道旧的kafka版本是提交给zookeeper的，但是新的kafka版本仍然是ZK，我甚至没有在配置文件中指明zookeeper的IP )。
(2)如果需要重复使用数据，如何检
浏览 4提问于2018-01-12得票数 0
使用从只有少数消息的主题中读取消息。卡夫卡一遍又一遍地重复队列中的信息。
它接收来自我的主题的一条消息(返回到每个消息内容)，然后抛出ERROR - Heartbeat session expired - marking coordinator dead并继续遍历其余的消息，并继续重播它们。更多的日志：
kafka.coordinator - ERROR - Heartbeat session expired - marking coordinator dead
kafka.coordinator - WARNING - Marking the coordinator dead (node 1
浏览 15提问于2017-09-29得票数 2
我有一个kafka消费者，如下所示： import akka.actor.ActorSystem
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.seria
浏览 64提问于2019-04-21得票数 1
2回答
我正在看如何记录所有卡夫卡主题/事件。有什么env变量可以运行我的Kafka容器吗？为了开发目的，我用了
docker run -p 9092:9092 -e ADVERTISED_HOST=127.0.0.1 johnnypark/kafka-zookeeper
当我运行python脚本时，我能够使用和发布事件。
from kafka import KafkaConsumer
consumer = KafkaConsumer('other-topic')
for msg in consumer:
print (msg)
from kafka import Kafk
浏览 9提问于2021-06-18得票数 1
回答已采纳
我正在尝试使用GetKafka处理器获取消息，但我无法获取。我使用Kafka命令行消费者测试了消费消息，它起作用了。我还能够成功地使用PutKafka processor将消息放入主题中。附加了我设置Zookeeper连接字符串和主题名称的设置。当我运行流时，我在处理器中看不到任何错误。
我在nifi-app.log中看到一个异常：
2016-08-03 09:34:33,722 WARN [70e1df87-6097-4ed0-9a40-7e36f9be6921_mydomain.com-1470231250839-1fbd0cfe-leader-finder-thread] kafka.
浏览 1提问于2016-08-03得票数 0
我们正在尝试优化我们的python应用程序，它通过confluent python kafka客户端生成消息到kafka主题，使用confluent控制中心，我们可以检查特定的消息是否传递到kafka主题，所以现在的问题是，我们应该在producing之后使用poll()函数，因为交付回调现在被删除了，如果我们在1秒内产生100K消息，它是否会导致更好的吞吐量。 producer.produce( topic=topicName, value=msg, key=msg_key)
producer.poll(0) # Is poll now required 我们使用的是在内部使用librdk
浏览 18提问于2020-11-10得票数 0
1回答
我是Kafka的新手，我正在使用Kafka 1.0。
我使用pull模式阅读Kafka消息，也就是说，我会定期poll() kafka主题以获取新消息，但我没有将偏移量写回Kafka。
我会问kafka如何知道我消耗了哪些偏移量，或者Kafka记住进度的机制是什么(Kafka偏移量)
浏览 9提问于2020-05-20得票数 1
回答已采纳
1回答
我想写信给消费者，它可以同时进行多个记录的投票。机制应该是这样运作的：
Poll 50条记录为list，如果任何记录不被复制，则做些什么，再次处理所有50条记录。我的当前配置：
spring:
cloud.stream:
kafka.streams.binder.brokers: ${bootstrap-server}
default.content-type: application/json
function.definition: functionName
bindings:
func
浏览 3提问于2022-03-21得票数 0
1回答
我们使用，我们的应用程序使用一个主题并处理消息，然后将它们输出到不同的主题。
这些主题也在同一个应用程序中使用，并输出到最后一个主题。
我们注意到，每当第一个使用者使用新消息而这些线程仍然处于活动状态时，就会创建大量的生产者线程。
以下是我的简化配置：
cloud:
stream:
function:
definition: schedulingConsumer;consumerSearch1;consumerSearch2
default:
group: ${kafka.group}
contentType:
浏览 3提问于2022-06-27得票数 1
回答已采纳
3回答
我在客户端使用kafka : kafka_2.12-2.1.0，spring kafka，遇到了一个问题。
我需要通过读取kafka主题中的所有现有消息来加载内存中的映射。为此，我启动了一个新的消费者(具有唯一的消费者组id，并将偏移量设置为earliest)。然后，我遍历消费者(poll方法)以获取所有消息，并在消费者记录为空时停止。
但我注意到，当我开始轮询时，前几次迭代返回的消费者记录为空，然后开始返回实际记录。现在，这打破了我的逻辑，因为我们的代码认为主题中没有记录。
我尝试了几种其他方法(比如使用偏移量)，但没有想出任何解决方案，除了在某个地方保留另一个记录，告诉我在我停止之前需要阅
浏览 1提问于2019-02-11得票数 3
我正在编写一个消费者应用程序，从kafka流中提取记录并使用spring进行处理。我的处理步骤如下：
Getting records from stream --> dump it into a table --> Fetch records and call API --> API will update records into a table --> calling Async Commit()
在某些情况下，API处理需要花费更多的时间，因为更多的记录正在被获取，而我们正在获得以下的错误？
由于使用者轮询超时，consumer-prov-em-1-399ede4
浏览 6提问于2020-10-15得票数 3
消息键不被传播，到达接收器的消息都用键等于NULL值到达好吗我做错了什么？
以下是实施的步骤：
来自protobuf格式主题：
打印'protobuf-topic';
键格式:跳转( KAFKA_STRING )或滚动( KAFKA_STRING )或KAFKA_STRING值格式: PROTOBUF或KAFKA_STRING行时: 2020/11/19 16:41:24.926 Z，键:KAFKA_STRING值: content："Hello world1“date_time："2020-11-19T16:41:23.932842Z”id：“375c71e9
浏览 5提问于2020-11-19得票数 0
3回答
我最近正在学习Kafka，我的消费者不能使用任何记录，除非我指定了--parititon 0参数。换句话说，我不能使用这样的记录：
kafka-console-consumer --bootstrap-server 127.0.0.10:9092 --topic first-topic
但其工作方式如下：
kafka-console-consumer --bootstrap-server 127.0.0.10:9092 --topic first-topic --partition 0
主要问题是，当我转到java代码时，我的KafkaConsumer类无法获取记录，我需要知道如何在java
浏览 5提问于2021-11-09得票数 1
回答已采纳
1回答
Kafka消费者未收到消费者启动前产生的消息。 public class MyKafkaConsumer {
private final KafkaConsumer<String, String> consumer;
private final String TOPIC="javaapp";
private final String BOOTSTRAP_SERVERS="localhost:9092";
private int receivedCounter=0;
private ExecutorServi
浏览 26提问于2020-04-12得票数 0
假设我有一个消费者正在运行，它正在从10个分区中获取数据。在一次轮询请求中，消费者为每个分区提取10条记录，因此总共提取100条记录。
现在，在向组中再添加一个消费者并重新平衡之后，两个消费者都从5个分区获取数据，每个消费者现在总共获取50条记录(每个分区10条记录)。
我想知道是否有一种方法可以配置Kafka Consumer，这样即使在添加了一个消费者之后，两个消费者也会开始为每个分区获取20条记录，因此总记录保持为100条。
我试着使用max.poll.records和fetch.max.bytes，但对我不起作用。在将fetch.max.bytes设置为1000之后，kafka从分区中
浏览 2提问于2020-09-29得票数 1
我轮询来自多个分区的记录，并且只提交一个记录，然后其余的记录似乎也被提交了，因为我不能再次轮询它们。并且我已经将auto.commit设置为false，但找不到原因。但是，如果该主题只有一个分区，它将正常工作，我在第一次轮询时没有提交的记录将再次重新轮询。当有多个分区时，为什么结果会不同？
public static void main(String[] args) throws IOException,
InterruptedException {
initKafkaConsumer();
do {
Consumer
浏览 15提问于2019-08-08得票数 1
1回答
spring.kafka.consumer.max-poll-records = 2000 //each record of size 5kb takes 100 ms so to process entire batch takes 500 sec i.e 8 min 20 sec
spring.kafka.consumer.properties.max.poll.interval.ms = 900000 //15 min
spring.kafka.consumer.properties.fetch.max.wait.ms = 600000 //10 min
spring.kafka.con
浏览 3提问于2021-07-01得票数 0
2回答
我有一个Kafka流应用程序，它从几个主题中获取数据，并将数据加入到另一个主题中。
Kafka配置：
5 kafka brokers
Kafka Topics - 15 partitions and 3 replication factor.
注:我在卡夫卡经纪人运行的同一台机器上运行卡夫卡流应用程序。
每小时消耗/产生数以百万计的记录。每当我把任何卡夫卡经纪人打倒，它就进入再平衡，它需要大约。30分钟，有时甚至更长的时间再平衡。
有人知道如何解决卡夫卡消费者的再平衡问题吗？而且，它多次在重新平衡的同时抛出异常。
这将阻止我们在生产环境中使用此设置。任何帮助都将不胜感激。
Caused by
浏览 3提问于2017-09-09得票数 6
回答已采纳
max.poll.records最近在kafka的消费者配置中被更改为500，但我想知道这对消费者调查有什么影响。它只是可以获取的最大记录数的上限，还是消费者等到它获得500条记录时才能获取。
浏览 1提问于2018-11-15得票数 5
回答已采纳
我有一个kafka使用者处理一个消息主题名为"ABC“，它具有配置：
{ org.apache.kafka.common.serialization.StringDeserializer，value.deserializer=class org.apache.kafka.common.serialization.StringDeserializer，max.poll.records=250，group.instance.id=ABC，group.id=ABC，bootstrap.servers=localhost:9093，auto.commit.interval.ms=50，secu
浏览 8提问于2020-06-15得票数 1
3回答
我正在尝试使用spring cloud数据流从oracle DB捕获更改数据，并使用kafka作为代理。为此，我使用了轮询机制。我使用基本的select查询定期轮询数据库，以捕获任何更新的数据。为了获得更好的防故障系统，我将上次轮询时间保存在oracle DB中，并使用它来获取在上次轮询后更新的数据。 public MessageSource<Object> jdbcMessageSource() {
JdbcPollingChannelAdapter jdbcPollingChannelAdapter =
new JdbcPollingChann
浏览 30提问于2019-03-07得票数 2
回答已采纳
2回答
我在一个Kafka使用者上有以下spark /Streaming，当批大小达到特定大小N时，如何指定获取应该是有条件的，否则使用者应该在处理之前缓冲元素，所以每当我想要执行我的逻辑时，都可以保证我有一个大小为N的Dataset<VideoEventData>：
Dataset<VideoEventData> ds = spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", prop.getProperty("
浏览 3提问于2017-11-27得票数 1
回答已采纳
我刚刚开始与卡夫卡合作，并开始在我的本地机器上玩卡夫卡。我能够使用confluent提供的生成和使用消息。
目前为止我的理解是，
kafka客户端(即消费者和生产者)维护一个消息队列。
生产者将把每条产生的消息存储在本地缓冲队列中。生产者需要显式地将消息从本地缓冲区推送到kafka集群。
在消费者端，它将以某种方式从kafka集群中获取消息，并将其存储在本地缓冲区队列中，然后通过api调用(如poll() )从缓冲区中提取这些消息
首先，我失踪了吗？
此外，我经常在两个客户端遇到一个短语“交换消息/抵消”。向卡夫卡传递信息/抵消究竟意味着什么？
浏览 0提问于2020-05-08得票数 0
回答已采纳
1回答
我在一个物理服务器上运行Zoo门将和Kafka，我构建了一个码头容器，运行一个java应用程序，作为使用者连接到外部Kafka。
卡夫卡监听端口：
[helio@dev-01 kafka]$ netstat -an|grep 9092
tcp6 0 0 :::9092 :::* LISTEN
我正在使用docker共享网络，但是消费者应用程序无法连接到kafka。我可以从集装箱里找到卡夫卡IP .我尝试了一个简单的方法，以确保我可以通过网络访问kafka服务器：
在集装箱：
root@ff419d415
浏览 3提问于2018-05-21得票数 0
回答已采纳
Apache Kafka文档指出：
内部Kafka流消费者max.poll.interval.ms默认值从300000改为Integer.MAX_VALUE。
由于此值用于检测一批记录的处理时间超过给定阈值的时间，是否存在这样一个“无限”值的原因？
它是否使应用程序变得没有响应能力？或者，卡夫卡流有一种不同的方式离开消费者群体时，处理过程太长？
浏览 1提问于2017-12-20得票数 12
回答已采纳
我试图在每次调用消费者时重置消费者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
我正在构建一个非常简单的KafkaStreams演示应用程序，以测试一个用例。
我无法升级我正在使用的Kafka broker (目前版本为0.10.0)，而且有几条消息是由一个预-0.10.0生成程序编写的，所以我使用了一个自定义TimestampExtractor，在我的主类开始时将它作为默认配置添加到配置中：
config.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, GenericRecordTimestampExtractor.class);
当使用我的源主题时，这是非常好的。但是在使用聚合操作符时，我遇到了一
浏览 0提问于2018-01-24得票数 8
回答已采纳
在我的Spring Boot Kafka应用程序中，我有以下使用者配置：
@Bean
public ConsumerFactory<String, Post> postConsumerFactory(KafkaProperties kafkaProperties) {
return new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties(), new StringDeserializer(), new JsonDeserializer<>(Post.class
浏览 2提问于2018-08-02得票数 4
回答已采纳
我们有一个基于Sprring的Kaffka消费者，我们为其创建了一个工厂，如下所示：
@Bean
public ConsumerFactory<String, Customer> customerConsumerFactory() {
Map<String, Object> config = new HashMap<>();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "bootstrapServers");
config.p
浏览 1提问于2021-03-08得票数 0
我们有一些长期运行的工作，也是用和Kafka绑定实现的。我们面临的问题是，因为max.poll.interval.ms和max.poll.records的默认值不适合我们的用例，我们需要为max.poll.interval.ms设置一个相对较大的值(几个小时)，为max.poll.records设置一个相对较小的值(例如，1)，以便与运行时间最长的作业保持一致。这解决了消费者进入再平衡循环的问题。然而，它给消费者带来了一些操作上的挑战。有时，使用者在重新启动时会被卡住，直到max.poll.interval.ms通过时才会使用任何消息。
这是因为Spring流轮询的实现方式吗？如果我使用同步使
浏览 4提问于2020-05-07得票数 0
回答已采纳
2回答
我使用的是卡夫卡连接器汇合3.0.1版本，我创建了一个名为 new - group 的新组，it.most上大约有20个主题是busy.But。遗憾的是，当我启动连接器框架时，系统无法停止再平衡，所有主题的再平衡时间大约为2分钟。我不知道原因。其中一些错误消息是：
[2017-01-03 21:43:57,718] ERROR Commit of WorkerSinkTask{id=new-connector-0} offsets threw an unexpected exception: (org.apache.kafka.connect.runtime.WorkerSinkTask:1
浏览 0提问于2017-01-03得票数 7
2回答
我怎么能从卡夫卡那里读得更快
、、、、
我创建了一个新的Kafka服务器(我创建了一个带有一个分区的代理)，我成功地使用了java代码在这个服务器上生成和使用，但我并不满足于我作为消费者每秒读取的事件数量。
我已经使用了下列消费者设置：
AUTO_OFFSET_RESET_CONFIG = "earliest"
FETCH_MAX_BYTES_CONFIG = 52428800
MAX_PARTITION_FETCH_BYTES_CONFIG = 1048576
MAX_POLL_RECORDS_CONFIG = 10000
pollDuration = 3000
但是，无论我向每个设置输入了什么值，结果都是相同的。
浏览 0提问于2019-08-08得票数 2
回答已采纳
我使用的是kafka消费者api 0.10.2.1。
KafkaConsumer为分区分配和撤消提供了一个回调：
consumer.subscribe(topics, consumerRebalanceListener);
其中consumerRebalanceListener有两个方法：
public void onPartitionsRevoked(Collection<TopicPartition> partitions);
public void onPartitionsAssigned(Collection<TopicPartition> partitions
浏览 6提问于2017-10-18得票数 2
回答已采纳
我用春天的卡夫卡来消费卡夫卡的信息。使用者侦听器如下所示。
@KafkaListener(topics = "topicName",
groupId = "groupId",
containerFactory = "kafkaListenerFactory")
public void onMessage(ConsumerRecord record) {
logger.info("Received Message from kafka topic " + reco
浏览 45提问于2022-05-27得票数 1
我们使用Spring Kafka 2.2.2版本通过@KafkaListener和ConcurrentKafkaListenerContainerFactory从Kafka中检索记录。我们已经将max-poll-records配置为5，但是它始终只向消费者提供列表中的1条记录，而不是5条记录。
虽然使用相同的配置，但它可以在Spring Kafka 2.1.4版本中工作。
下面是我们的application.yml配置：
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
enable-auto-
浏览 0提问于2019-06-23得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
