Spring Cloud Stream Kafka批量-手动提交整个批量

文章/答案/技术大牛

发布

2回答

spring-kafka、spring-cloud-stream

我们使用spring cloud streams Hoxton.SR4来消费来自Kafka topic的消息。我们已经启用了每次轮询获取2000条记录的spring.cloud.stream.bindings..consumer.batch-mode=true，。我想知道是否有一种方法可以手动确认/提交整个批次。

浏览 264提问于2020-12-10得票数 0

1回答

SpringCloudStream Kafka如何批量消费？

apache-kafka、apache-kafka-streams、spring-kafka

可以使用原生Kafka KafkaConsumer.poll ()批量获取消息。集成spring cloud starter stream Kafka (3.0.10.发布)？

浏览 56提问于2020-12-17得票数 0

1回答

Spring Cloud Stream -批量接收消息

spring-cloud-stream、spring-cloud-dataflow、spring-kafka

我希望以固定的时间间隔从Kafka主题获取一批消息(比如10000条消息)(用于一次性批量处理所有10k条记录，而不是逐条处理)。有没有办法在Spring Cloud Stream processor中实现这一点？如果是这样，有没有我可以参考的代码片段或示例？谢谢

浏览 14提问于2017-12-15得票数 1

回答已采纳

1回答

在Spring中通过Kafka消耗批处理

java、spring-boot、maven、apache-kafka、spring-cloud-stream

我是卡夫卡的新手，想要通过消费者批量加工。 <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-

浏览 3提问于2020-07-08得票数 1

回答已采纳

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.<em

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

回答已采纳

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，但得到上述异常。=TOPIC.FEED.NAMEspring.cloud.stream.bindin

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

回答已采纳

1回答

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

spring-kafka、spring-cloud-stream

: destination: consumer.first group: ${spring.application.namesecond-consumer-in-0: use-native-encoding: true group: ${spring.application.name但是，当我的spring引导应用程序启动时，我看到了两个消

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

1回答

向kafka* dlq发送错误消息时获取ClassCastException*

java、apache-kafka、kafka-consumer-api、spring-cloud-stream

我使用的是kafka和spring cloud streams。is in unnamed module of loader 'app') at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.lambda$getErrorMessageHandler$8(KafkaMessageChannelBinder.java:1057) ~[spring-

浏览 158提问于2020-04-03得票数 2

1回答

春云探测不同轨迹ID与卡夫卡的集成

spring、spring-boot、spring-cloud、spring-cloud-sleuth

我使用Kafka实现微服务之间的Asyng调用，使用Spring进行日志记录。

浏览 2提问于2016-12-02得票数 4

回答已采纳

1回答

如果我们在Kafka监听器中使用return语句会发生什么？

java、spring、spring-boot、apache-kafka、spring-kafka

我有一个可以批量消费消息的Kafka监听器。我也在使用手动提交。如果其中任何一条消息出现问题，我将使用返回语句，而不是提交偏移量(我希望再次处理所有这些消息)。我的问题是- 在Kafka Listener中使用return语句有副作用吗？再次消费相同的消息需要多长时间？我正在使用Spring Kafka和SpringBoot。

浏览 34提问于2021-04-26得票数 0

回答已采纳

2回答

Spring云流:标头kafka_acknowledgment值为空

spring-boot、spring-integration、spring-cloud-stream、spring-kafka

根据应用程序中消息处理的状态，我试图控制对卡夫卡主题的偏移提交。如果消息成功，则可以提交偏移量。为此，我正在尝试获取方法中的标头，如果是这样，则可以手动确认该消息。spring: stream: contentType: application/json type: kafka defaultCandid

浏览 0提问于2018-07-03得票数 0

回答已采纳

2回答

从kafka中的投票记录创建批次

java、apache-kafka、apache-kafka-connect

我们需要对elasticsearch执行批量写入。我们想知道是否有更好的方法对数据进行批处理，并在进行批处理时避免数据丢失。initialize(topic, consGroup, STREAMSERDE.STRINGDESER, STREAMSERDE.JSONDESER, props); logger.debug("Kafka

浏览 1提问于2019-01-22得票数 0

1回答

春季云流手册Poller Kafka

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

我不想使用@KafkaListener或@StreamListener，但我想手动投票kafka。poll.forEach(record -> { });application.propertiesspring.

浏览 1提问于2020-02-05得票数 0

2回答

用Spring总线刷新配置更改

java、spring-boot、apache-kafka、spring-boot-actuator

但是，我没有手动刷新每个客户端应用程序以在提交后提取属性文件中的最新更改，而是使用Spring和Kafka作为消息代理，以便将所有更改广播给客户端应用程序。下面是pom文件和属性文件。<dependency> <artifactId>spring-cloud-starter-b

浏览 3提问于2019-01-29得票数 3

回答已采纳

1回答

autoCommit是如何工作的，在spring-cloud-stream反应式kafka中会有多少消息被轮询一次？

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

我查看了document，它只说明了自动提交如何与poll() here一起工作，以及如何配置轮询计数here。那么当我使用Flux时，事情是如何工作的呢？下面是我的消费者代码。

浏览 9提问于2020-12-07得票数 0

回答已采纳

2回答

用弹簧云流实现kafka的自动提交

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

我有一个应用程序，我想要在卡夫卡消息中手动执行(n)ack。根据spring云文档，它应该使用autoCommitOffset 来完成。下面是我的配置spring.cloud.stream.default.contentType=application/json spring

浏览 7提问于2020-05-14得票数 1

回答已采纳

1回答

如何配置DeadLetterPublisherRecoverer在Stream批处理模式下向DLQ发送错误消息

java、spring、spring-kafka、spring-cloud-stream、spring-cloud-function

我使用和Spring函数创建了一个Kafka消费者，用于批量消费来自Kafka主题的消息。现在，我希望将错误批次发送到死信队列中，以便进一步调试错误。我正在用Spring重试处理我的使用者方法中的重试。但是对于不可重试的异常，我希望将整个批处理发送到DLQ。),} 但是这并不是向error.topic发送任何消息，从错误日志中我可以看到它试图连接到localhost:9092，而不是我在spring.<

浏览 9提问于2022-07-11得票数 0

1回答

单记录kafka消费者和kafka批量消费者的基本区别是什么？

spring-kafka

我正在使用spring-kafka 2.2.8，试图理解单个记录消费者和批量消费者之间的主要区别是什么。据我所知，从一个主题中读取消息/字节对于单个记录消费者和批量消费者来说没有什么不同。唯一的区别是如何提交偏移量。并因此进行错误处理。我的理解正确吗？请确认。

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

回答已采纳

1回答

Spring Boot无法启动，因为使用@EnableBinding和zkNodes不可用

spring-cloud-stream

我有使用spring.cloud.stream.kafka.binder.brokers=KAFKA_URL spring.cloud.stream.kafka.binder.zkNodes=ZKNODE_URL的消费者如果我尝试运行spring boot消费者应用程序，但bean未启动，则整个spring boot应用程序无法启动: rg.springframework.con

浏览 4提问于2017-03-21得票数 0

1回答

消息的手动确认(检查点)：Spring* Cloud Stream Kenesis绑定器*

spring-cloud-stream、spring-cloud-dataflow

我们正在尝试将我们的Spring Cloud Stream应用程序从Kafka迁移到AWS Kenesis。我们需要手动确认来处理某些超时情况。对于Kafka，我们使用属性autocommitoffset设置为false，并使用确认头来处理手动确认。我看了一下Spring Cloud Stream的文档，看了下面的：但是找不到任何解决方案。任何指针都会非常有帮助。

浏览 31提问于2021-03-05得票数 0

回答已采纳

点击加载更多