Kafka Consumer是否默认批量大小？

文章/答案/技术大牛

发布

1回答

java、apache-kafka

Kafka是否提供了从topic读取消息的默认批量大小？我有以下从主题中读取消息的代码。while (true) { consumer.poll= record.value(); }); insertInBatch(user

浏览 91提问于2019-12-13得票数 3

1回答

如何将蔚蓝服务总线配置为批量发送消息？

java、apache-camel、apache-camel-k

可以配置kafka消费者，并在调试时确认其在每次轮询kafka主题时消耗数千条消息。 from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{kafka.port}}" + &

浏览 11提问于2022-06-26得票数 0

0回答

Confluent Kafka Python库为批量消息配置生产者

python、apache-kafka、kafka-producer-api、confluent-platform

我需要设置Kafka生产者发送一批500个味精，而不是味精，而是批量导入。TypeError: function takes at most 8 arguments (501 given)from confluent_kafka import Consumer, Producer producer = Producer(**{'bootstrap.s

浏览 12提问于2017-06-15得票数 0

回答已采纳

1回答

spring cloud stream kafka batch不会每15分钟消费一次消息，即使增加了这个配置，'fetch.max.wait.ms‘

spring-boot、apache-kafka、spring-cloud-stream、consumer、spring-cloud-stream-binder-kafka

我想每15分钟批量消费一次消息。为此，我设置了这些属性， spring.cloud.stream.kafka.binder.consumer-properties.max.poll.records=5000000spring.cloud.stream.kafka.binder.consumer-properties

浏览 131提问于2021-09-22得票数 0

回答已采纳

1回答

我们可以在一个应用程序属性文件中拥有两个不同的spring kafka消费者属性吗？

spring-kafka

Consumer1配置详情(批量)spring.kafka.consumer.[0].enable-auto-commit = falsespring.kafka.consumer</

浏览 9提问于2021-05-19得票数 0

回答已采纳

1回答

我应该在什么情况下使用批处理使用者与单记录使用者

spring-kafka

据我所知，Apache Kafka文档中没有诸如批处理消费者之类的特殊概念，但spring-kafka可以选择使用以下代码片段创建批处理消费者。factory.setBatchListener(true); // <<<<<<<<<<<<<<<<<<<<<<<<<} 现在我的问题是，我应该在什么时候使用批量消费者与单记录消费者有

浏览 17提问于2020-10-21得票数 0

回答已采纳

3回答

如何使用Spring Boot等待完整的Kafka消息批？

spring、spring-boot、kotlin、apache-kafka、spring-kafka

批量消费Kafka消息时，可以使用max.poll.records限制批量大小。如果消费者非常快，并且其提交偏移量没有明显滞后，这意味着大多数批处理将小得多。我希望只接收“完整”批处理，也就是说，只有在达到批处理大小时才调用我的消费者函数。所以我正在寻找像min.poll.records这样的东西，它并不以这种形式存在。下面是我正在做的一个最小的例子： import org.apache.kafka.clients.consumer.ConsumerConfig im

浏览 64提问于2020-01-02得票数 2

回答已采纳

1回答

aiokafka:考虑kafka消费群组再均衡时，消息处理是否需要加锁

python、apache-kafka、aio

当使用ConsumerRebalanceListener启动aiokafka消费者时，我们对从kafka接收到的批量消息(getmany())有一个处理过程。实际上，我不太确定在这种情况下我们是否真的需要锁，所以如果这里的人们能给出一些建议，我将不胜感激。虽然我们在这种情况下使用aiokafka，但我猜这对Kafka来说可能是一个一般性的问题。= asyncio.Lock() self.kafka_consumer = AIOKafkaConsume

浏览 53提问于2020-11-08得票数 0

1回答

春云流卡夫卡绑定，如何为两个消费者设置不同的max.polls.records设置？

spring-kafka、spring-cloud-stream

application.yml first-consumer-in-0: use-native-encoding: true content-type: application/*+avro destination: consumer.second use-native-enco

浏览 9提问于2022-08-05得票数 0

1回答

Python Kafka使用者在消息到达时不读取消息

python、apache-kafka、kafka-consumer-api

我刚刚开始使用Kafka，kafka-python。在下面的代码中，我尝试在消息到达时读取它们。但由于某些原因，消费者似乎在等待一定数量的消息积累后才获取它们。我最初认为这是因为制作人正在批量出版。当我运行"kafka-console-consumer --bootstrap-servers --topic“时，我可以在消息发布后立即看到收到的每条消息(就像在消费者控制台上看到的那样)。() pr

浏览 58提问于2019-12-21得票数 2

1回答

获取异常：[使用批处理后无法将B强制转换为org.springframework.messaging.Message异常

java、spring、apache-kafka、spring-cloud-stream、spring-cloud-stream-binder-kafka

我使用spring-cloud-stream-kafka-binder-3.0.4批量消费消息，消费后将消息转换为object，但得到上述异常。=truespring.cloud.stream.kafka.bindings.activity-input-channel.consumer.auto-commit-offset=false spring.cl

浏览 64提问于2021-09-17得票数 0

回答已采纳

1回答

错误:字段Message.Field .protobuf.MessageTypeAck.sourceModuleID的非法导线类型:1(预期为0)

javascript、node.js、protocol-buffers

我有一个应用程序，用kafka和协议缓冲区来处理和消费消息，一切都很好。我正在用SerializeAsString()序列化协议缓冲区(这个应用程序是用c++编写的)。) {} consumer.on

浏览 1提问于2016-05-01得票数 3

回答已采纳

2回答

最后，python代码没有运行

python

def kafka_messages(topic, partition): consumer = SimpleConsumer(kafka, b"consumer-group" consumer.provide_pa

浏览 2提问于2014-11-03得票数 0

回答已采纳

1回答

如何在命令行中使用带有人类可读的时间戳的Kafka消息？

apache-kafka

我知道kafka-console-consumer有一个属性print.timestamp=true，但是它打印的是划时代的时间，而不是人类可读的时间。如何在命令行中使用人类可读的时间戳来使用Kafka消息？

浏览 1提问于2019-10-09得票数 1

1回答

如何使用${akka.kafka.consumer}？

java、akka

我正在尝试使用默认的Kafka配置设置，但我不确定${akka.kafka.consumer}是如何设置的。阅读我定义了以下内容：our-kafka-consumer: ${akka.kafka.consumer} { }但是，当

浏览 18提问于2022-02-28得票数 0

回答已采纳

4回答

使用TPL数据流对持续时间或阈值进行批处理

c#、task-parallel-library、tpl-dataflow、batching

我已经使用TPL数据流实现了一个producer..consumer模式。用例是代码从Kafka总线读取消息。为了提高效率，我们需要在访问数据库时批量处理消息。在TPL数据流中，有没有一种方法可以在达到大小或持续时间阈值时保持消息并触发？例如，一旦消息从队列中拉出，当前实现就会发送该消息。

浏览 4提问于2018-10-04得票数 11

2回答

Python Kafka模仿返回类型的Kafka Consumer

python、unit-testing、apache-kafka、mocking

当我使用kafka-python包时，我想测试一个脚本。我想测试函数的返回对象类型 consumer = KafkaConsumer(bootstrap_servers=_KAFKA_BOOTSTRAP_SERVICE, consumer.su

浏览 20提问于2020-07-30得票数 1

1回答

Kafka-node突然从偏移量0开始消耗

node.js、apache-kafka、kafka-consumer-api、node-kafka

有时，kafka-node消费者从偏移量0开始消费，而它的默认行为是只消费较新的消息。则它将不会切换回其默认行为。你知道如何解决这个问题吗?知道发生了什么，它的行为突然改变了吗？var kafka = require("kafka-node") ; client = new kafka.KafkaClient();consumer =

浏览 5提问于2020-01-31得票数 3

1回答

Kafka删除未使用的消费群体

apache-kafka、spring-kafka

(例如my_consumer_group_123324234234，my_consumer_group_123324234235 .)。我想知道是否可以通过脚本删除未使用的使用者组(使用“my_consumer_group_*”之类的名称进行过滤)

浏览 2提问于2022-05-19得票数 0

2回答

分区非常大的__consumer_offsets主题

apache-kafka

我使用的是Kafka 2.0.0，有一些__consumer_offsets主题的分区是500-700 GB，超过5000-7000个段。这些部分超过2-3个月。日志中没有错误，该主题默认是紧凑的。或者可能是kafka 2.0.0的一个bug？我能做些什么检查？

浏览 58提问于2021-06-03得票数 0

点击加载更多