尝试使用Spring Cloud Stream 3.0批量消费kafka消息。 消费者收到一个包含单个记录的列表，而不是更多记录。下面是使用的用户编码的yml spring: stream:
浏览 49提问于2020-01-05得票数 1
我希望以固定的时间间隔从Kafka主题获取一批消息(比如10000条消息)(用于一次性批量处理所有10k条记录，而不是逐条处理)。有没有办法在Spring Cloud Stream processor中实现这一点？如果是这样，有没有我可以参考的代码片段或示例？
谢谢
浏览 14提问于2017-12-15得票数 1
回答已采纳
我正在尝试将单独的Serde放在我的每个消费者身上，而不是所有消费者。例如，这会将Avro反序列化程序添加到我的应用程序中的所有使用者。我只想把它放在一个消费者身上 spring.cloud.stream.kafka.binder.consumerProperties.value.deserializer=io.confluent.kaf
浏览 13提问于2020-10-28得票数 0
回答已采纳
: destination: consumer.first group: ${spring.application.namepartitioned: true configuration.isolation.level: read_committed
我为第一个消费者和第二个消费者设置了不同的但是，当我的spri
浏览 9提问于2022-08-05得票数 0
使用spring-cloud-stream的kafka绑定器，如何配置并发消息消费者(在单个消费者jvm中)？Kafka文档提到了循环分区。
s-c-s文档根本没有提到spring.cloud.stream.bindings.*.concurrency，尽管这在我上面描述的用例中似乎很重要。spring
浏览 2提问于2016-03-08得票数 7
1回答
春云流-消费者组绑定
、、、、
我的消费者被绑定到匿名消费者组，而不是我指定的消费者组。spring: stream: binder: defaultBrokerPortdestination: event group: e
浏览 0提问于2019-02-24得票数 2
回答已采纳
我正在使用spring-kafka和spring-kafka-test版本1.0.2.RELEASE。在我的一个测试中，我的应用程序使用KafkaTemplate和大多数默认配置设置将100条记录连续发送到EmbeddedKafka实例上的单个TopicPartion。我使用KafkaTestUtils.getRecords(消费者)方法尝试从Kafka实例<em
浏览 3提问于2016-08-12得票数 2
1回答
我正在使用Alpakka kafka和scala应用程序。我的Kafka在docker内部运行，我正在尝试使用我的代码在Kafka producer上发布消息。case Success(v) => v Future.failed(e) }
代码没问题，但当我发送大的xml文件时，它会给出an error that org.apache.kafka</e
浏览 4提问于2020-07-17得票数 1
Kafka使用者有一个配置max.poll.records，它控制在一次调用poll()及其时返回的最大记录数。我将其设置为一个非常高的数字，这样我就可以在一次投票中获得所有消息。然而，尽管该主题有更多的消息，但该调查在一次调用中仅返回数千条消息(约6000条)。
如何进一步增加单个消费者读取的消息数？
浏览 0提问于2018-08-09得票数 31
回答已采纳
我想为我的配置服务器编写单元/集成测试，所以我检查了链接。里面有一个用@SpringBootTest注释的类和一个用@BeforeClass注释的函数startConfigServer()。我还使用了Spring Cloud Stream Kafka和Spring Cloud Bus，同时启用了Spring Security基本身份验证。
浏览 7提问于2018-07-09得票数 0
我正在构建一个spring云微服务，它使用来自kafka主题的数据。在消费者中，我将主题绑定到一个KStream。传入消息不包含时间戳，因为kafka版本低于0.10。下面是我的配置：
org.apache.kafka.common.serialization.Serdes$StringSerde : cloud: stream: org.apache.kafka.common.s
浏览 1提问于2019-06-08得票数 0
在启动Spring数据流流时，由于与Kafka相关的各种错误，应用程序常常无法在我的机器上部署。例如：org.springframework.cloud.stream.binder.BinderException: Cannot initialize binder [...]Caused by: kafka.common.KafkaException: fetching topic metadata for topics [S
浏览 2提问于2016-10-04得票数 2
回答已采纳
String>> process() {
}}
我的期望是这应该用"foo“替换每条消息，但它所做的只是复制生产者主题上的消息。我也尝试过使用Transformer，但它具有相同的行为。将消息实际转换为"foo“的最简单方法是什么？
浏览 0提问于2020-10-01得票数 2
我使用和Spring函数创建了一个Kafka消费者，用于批量消费来自Kafka主题的消息。现在，我希望将错误批次发送到死信队列中，以便进一步调试错误。我正在用Spring重试处理我的使用者方法中的重试。但是对于不可重试的异常，我希望将整个批处理发送到DLQ。这就是我的消费者的样子：public C
浏览 9提问于2022-07-11得票数 0
我在KafkaDirectStream下面创建。ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams)) val lines = messages.map(_.value)我如何实现我的目标？我在<
浏览 0提问于2018-06-09得票数 0
我们使用spring cloud streams Hoxton.SR4来消费来自Kafka topic的消息。我们已经启用了每次轮询获取2000条记录的spring.cloud.stream.bindings..consumer.batch-mode=true，。
浏览 264提问于2020-12-10得票数 0
我的消费者配置有kafka批量监听器配置和@KafkaListener消费消息列表。我有一个ConsumerInterceptor，我想为每条记录设置惟一的id，并将它的值存储在MDC中。如果我的kafka监听器消费了一条消息，那么唯一的id是正确的。但是我的kafka侦听器
浏览 14提问于2021-06-11得票数 0
我们使用Spring Kafka 2.2.2版本通过@KafkaListener和ConcurrentKafkaListenerContainerFactory从Kafka中检索记录。我们已经将max-poll-records配置为5，但是它始终只向消费者提供列表中的1条记录，而不是5条记录。
虽然使用相同的配置，但它可以在<e
浏览 0提问于2019-06-23得票数 1
考虑一下这个Kafka使用者，它接收来自主题的数据，将其缓冲到PreparedStatement中，并且当100 K记录被批处理时，它向DB发出插入查询。
在数据仍然输入之前，这是很好的工作方式。但是，当20K记录被缓冲，并且没有更多的记录进入时，它仍然等待更多的80K记录，直到in刷新语句。但是，，我想冲洗那些20K的，如果停了一段时间后。我怎么能这么做？我看不
浏览 1提问于2019-06-05得票数 2
回答已采纳
1回答
我正在努力将与Kafka绑定集成在一起。我的应用程序从主题中消耗json，并将其反序列化为Java对象。我使用的是功能风格方法，而不是祈使式。我的代码使用结构良好的json输入。另一方面，当我发送无效的json时，我希望触发错误日志记录方法。这在一些测试用例中有效，而在另一个测试用例中不起作用。我的应用程序反序列化json，即使它是无效的，并且触发包含逻辑的方法，
浏览 12提问于2021-06-02得票数 0
回答已采纳
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储
腾讯云开发者
