被覆盖的MessageListner的onMessage在Spring Kafka消费者单元测试中未被调用
在Spring Kafka消费者单元测试中，如果被覆盖的MessageListener的onMessage方法未被调用，可能是由于以下几个原因：
- 配置错误：检查消费者的配置是否正确。确保消费者连接到正确的Kafka集群，并且订阅了正确的主题。
- 消息未发送：确认是否有消息发送到了被测试的Kafka主题。可以通过在测试代码中手动发送消息来验证。
- 消费者未启动：确保消费者已经启动并且正在运行。在单元测试中，需要手动启动消费者并等待一段时间以确保消费者已经成功连接到Kafka集群。
- 消费者组错误：如果消费者使用了消费者组，确保消费者组的配置正确，并且消费者组中的其他消费者没有消费掉所有的消息。
- 消息过滤：检查消息过滤器是否正确配置。如果消息过滤器设置不正确，可能会导致某些消息被过滤掉而不被消费者处理。
- 异步处理：在某些情况下，消费者的消息处理可能是异步的。在单元测试中，需要适当地等待一段时间以确保消息已经被处理。
如果以上步骤都没有解决问题，可以考虑检查消费者的日志以查看是否有任何错误或异常信息。此外，还可以尝试使用调试工具来跟踪代码执行过程，以确定为什么被覆盖的MessageListener的onMessage方法未被调用。
腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CKafka：https://cloud.tencent.com/product/ckafka
- 腾讯云云原生容器服务 TKE：https://cloud.tencent.com/product/tke
- 腾讯云云服务器 CVM：https://cloud.tencent.com/product/cvm
- 腾讯云云数据库 MySQL：https://cloud.tencent.com/product/cdb_mysql
- 腾讯云云存储 COS：https://cloud.tencent.com/product/cos
- 腾讯云人工智能 AI：https://cloud.tencent.com/product/ai
- 腾讯云物联网平台 IoT Hub：https://cloud.tencent.com/product/iothub
- 腾讯云移动开发 MSDK：https://cloud.tencent.com/product/msdk
- 腾讯云区块链服务 TBC：https://cloud.tencent.com/product/tbc
- 腾讯云元宇宙服务：https://cloud.tencent.com/product/um
相关·内容
我有一个带有Kafka应用程序的Springboot，我的Kafka集群从docker文件开始。但是，如果Kafka集群保持启动，但是Spring应用程序重新启动，则无法接收任何消息(函数正常工作)。
属性文件：
server.port=18080
spring.kafka.bootstrap-servers=localhost:29092,localhost:39092
spring.kafka.listener.ack-mode=MANUAL_IMMEDIATE
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.cons
浏览 8提问于2022-05-31得票数 0
在Spring Boot项目中使用Spring for Apache Kafka时，有没有办法通过application.properties中的条目配置不同的侦听器 例如，如果我在主题foo和bar上有不同的监听程序，我是否可以有如下条目： spring.kafka.producer.foo.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.bar.value-serializer=some.other.Serializer 我想我在什么地方看
浏览 30提问于2020-03-29得票数 0
1回答
弹簧靴卡夫卡手册失败
、、、
我有一个Spring消费者的下面的配置。我在尝试手动确认而不是自动提交。在手动确认后，我开始出错。Spring版本为2.7.2。
kafka.consumer.groupId=mcs-ccp-event
message.topic.name=mcs_ccp_test
kafka.bootstrapAddress=kafka-dev-app-a1.com:9092
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.ack-mode=MANUAL_IMMEDIATE
和这种使用者配置
@EnableKafk
浏览 26提问于2022-08-06得票数 0
回答已采纳
1回答
我经历了以下两个环节：
它们都有两种不同的方法，最多只能实现一次消费者语义，所以我有点搞不懂该遵循哪一种方法。
我当前的使用者配置是：
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.bootstrap-servers=<server port>
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization
浏览 0提问于2020-08-17得票数 0
我正在尝试为我正在使用Spring Boot2.x开发的Kafka侦听器编写单元测试。作为一个单元测试，我不想启动一个完整的Kafka服务器，一个Zookeeper的实例。因此，我决定使用Spring嵌入式Kafka。
我的听众的定义是非常基本的。
@Component
public class Listener {
private final CountDownLatch latch;
@Autowired
public Listener(CountDownLatch latch) {
this.latch = latch;
}
@
浏览 0提问于2018-05-02得票数 20
回答已采纳
1回答
如何对实现spring接口的类进行单元测试？我有一个监听器类，它使用onMessage函数手动侦听主题。这个函数非常简单，只接收消息。
我的设置是Spring5.8，Spring-Kafka2.2.7，Spring，JUnit和，没有弹簧引导。
我已经尝试了许多来自Spring和其他帖子的不同示例，但似乎没有一种测试实现MessageListener的侦听器类的简单方法。
我不确定是否需要设置EmbeddedKafkaBroker或EmbeddedKafkaRule，或者是否有不同的测试方法。当我尝试使用EmbeddedKafkaRule时，我得到了一个错误，上面写着NoClassDefFou
浏览 0提问于2019-07-24得票数 0
回答已采纳
1回答
我们有一个非常简单的卡夫卡消费者(v2.6.2)。它是消费者组中的唯一使用者，而这个组是唯一一个从主题中读取信息的组(有6个分区，其中包含大约300万个事件)。代理也是2.6.x版本。
由于我们需要实现一个"excatly-once"场景，我们深入研究了一下，如果我们真的只使用了一次写入主题的事件。不幸的是，我们发现：--使用者有时跳过一个偏移量，有时甚至跳过一组分区的偏移量。
使用者所做的仅仅是日志记录。错误处理程序配置有日志记录。
@KafkaListener(id = "${spring.kafka.consumer.group-id}", topics
浏览 9提问于2022-02-28得票数 1
在使用SeekToCurrentErrorHandler之后，我正在寻找一个非阻塞的kafka ErrorHandler。由于某些子系统不稳定，我们需要将高间隔时间设置为5分钟或更长时间。这会阻碍我们的处理。
我的想法是使用主题本身对失败的消息进行重新排队。但是具有两个附加的报头值kafka_try-counter和kafka_try-timestamp。
在SeekToCurrentErrorHandler和DeadLetterPublishingRecoverer的基础上，实现了RePublishingErrorHandler和RePublishingRecoverer的草案
RePubl
浏览 1提问于2020-03-11得票数 0
使用该配置
使用spring-integration-kafka扩展和以下配置：
<int-kafka:zookeeper-connect id="zookeeperConnect"
zk-connect="#{kafkaConfig['zooKeeperUrl']}" zk-connection-timeout="10000"
zk-session-timeout="10000" zk-sync-time="2000" />
<int-kafka:consume
浏览 0提问于2016-01-29得票数 0
我正在尝试使用Kafka-Spring (spring-boot)实现一个Kafka消费者。我使用生产者发送了一条消息，但是Spring抛出了一个异常，并且由于这个异常，我的侦听器方法没有被调用。
下面是我的监听器方法：
@Service
public class KafkaExchangeApi {
@KafkaListener(topics = ORDER_TOPIC, groupId = "group_id")
public void handleOrderResponse(@Payload GatewayOrder response) {
浏览 0提问于2020-09-10得票数 1
1回答
我使用kafka和spring引导应用程序在阅读完主题消息后使用消息，我将收到此错误。
2020-07-07 17:05:32,265 INFO GroupCoordinator 1001:组员consumer-1-708fe639-905d-432a-b6ba-17bf58adcf03 -service-prod已经离开，从小组(kafka.coordinator.group.GroupCoordinator) 2020-07 17:05:32,267 INFO GroupCoordinator 1001:准备重新平衡团体自行车-服务-prod在州与老一代460 (_消费者_偏移-7)(理
浏览 3提问于2020-07-07得票数 2
回答已采纳
我在Spring应用程序中使用Spring。我试图使用卡夫卡ConsumerInterceptor拦截时，偏移是提交的。
这似乎起作用了，生产者事务没有启用，但是事务打开了，Interceptor::onCommit不再被调用。
下面的最小示例一切都如预期的那样工作：
@SpringBootApplication
@EnableKafka
class Application {
@KafkaListener(topics = ["test"])
fun onMessage(message: String) {
log.warn("onMe
浏览 6提问于2022-10-13得票数 2
回答已采纳
2回答
问题中提到了集群配置、类详细信息和jar版本。
我已经启动了动物园管理员服务器、卡夫卡服务器和卡夫卡REST服务器。接下来，我将在tomcat上部署名为spring-kafka-webhook-service.war文件的春季引导war文件。
当我通过Kafka代理客户端发布消息时，我得到了下面的错误，这可能表明@KafkaListener方法无法读取ConsumerRecord传入的消息。任何投入都将受到高度赞赏。
我的Kafka-Rest属性当前配置如下：
confluent-3.3.0/etc/kafka-rest/kafka-rest.properties
id=kafka-rest
浏览 6提问于2017-12-26得票数 1
回答已采纳
我有一个Spring Boot应用程序，它监听Kafka队列上的消息。要过滤这些消息，请使用以下两个类
@Component
public class Listener implements MessageListener {
private final CountDownLatch latch1 = new CountDownLatch(1);
@Override
@KafkaListener(topics = "${spring.kafka.topic.boot}")
public void onMessage(Object o) {
浏览 6提问于2018-02-18得票数 3
我想用ack模式MANUAL_IMMDEDIATE实现手动提交。我使用AcknowledgingMessageListener类作为消费者。它在应用程序运行时很好，但是在用户接收数据之后，应用程序就会出现错误，但是应用程序仍然在运行。只有在收到数据后才会发生错误。任何建议都将不胜感激，谢谢
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.co
浏览 37提问于2022-06-07得票数 1
回答已采纳
使用spring- Dalston.SR4运行spring引导的Dalston.SR4程序。我通过云流进行了研究，特别是关于kafka绑定器的使用，但在一些配置方面还不清楚：
当我们从卡夫卡经纪人那里消费时，我们真的需要设置spring.cloud.stream.{instanceCount,instanceIndex}吗？考虑到设置了spring.cloud.stream.bindings.<channel>.group，我不明白为什么需要这样做。还是说只有在cloud.stream.kafka.bindings.<channel>.autoRebalanceEn
浏览 2提问于2017-10-13得票数 0
回答已采纳
1回答
在卡夫卡消费者方面，我一直面临以下例外。令人惊讶的是，这个问题是不一致的和代码的旧版本(配置完全相同，但一些新的无关特性)工作正常。有人能帮我确定是什么原因造成的吗？
[ERROR][938f3c68-f481-4224-b2c6-43af5fb27ada-0-C-1][org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer] - Error handler threw an exception
org.springframework.kafka.KafkaException: Seek
浏览 8提问于2021-06-11得票数 0
回答已采纳
1回答
我已经注意到，经过一段时间的-for示例，两天后，消费组并发性比I配置更低。
我使用spring引导，下面是我的代码示例
factory.setConcurrency(10);
当我在说明kafka使用者之后使用kafka命令时，它正好显示了10个不同的使用者客户端。
bin/kafka-consumer-groups.sh --describe --group samplaConsumer --bootstrap-server localhost:9092
经过一段时间后，当我运行上级命令时，使用者客户端变得更低，例如6个不同的客户端并管理这10个分区。
在再平衡或任何数量的客户保持不变之后
浏览 0提问于2019-08-31得票数 1
回答已采纳
1回答
我创建了一个单元测试来测试Kafka侦听器，如下所示。
@SpringBootTest
@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092","port=909"})
class ConsumerTest {
@Autowired
KafkaTemplate producer;
@Test
public void consumeEvents1Test() throws InterruptedExcept
浏览 0提问于2021-02-24得票数 0
回答已采纳
我想知道spring boot内部的spring kafka Producer是否保证交付。
有人知道如果某个随机监听器无法接收消息会发生什么情况吗？spring kafka是否会重试发送消息？
浏览 0提问于2019-05-29得票数 0
我是卡夫卡的新手，我想要帮助
我有两个应用程序(springboot)，它们是相同的/拷贝只有不同的端口。
他们都是消费者
这两个人听XXX这个话题
我有另一个应用程序扮演制片人的角色。
每当我给某某话题发东西。
只有其中一个使用消息，而另一个不使用消息。
我分别测试了两种方法，如果他们是单独听的话，他们也会正常地听，但是如果他们在一起听，只有一人会听。
我在用
<dependency>
<groupId>org.springframework.cloud</groupId>
<a
浏览 0提问于2021-07-06得票数 1
回答已采纳
由于不触发SleuthKafkaAspect.wrapProducerFactory()方法，跟踪信息不会在kafka消息上传播。在生产者端，消息被正确发送，跟踪信息被正确记录。在消费者方面，反而创建了一个新的traceId和spanId。
以下两条日志记录行显示了traceId、spanId (和parentId)的不同值：
2021-03-23 11:42:30.158 [http-nio-9185-exec-2] INFO my.company.Producer - /4afe07273872918b/4afe07273872918b// - Sending event='My
浏览 4提问于2021-03-23得票数 0
回答已采纳
我有一个使用Spring Cloud Stream编写的Kafka消费者，它连接到部署在Amazon MSK集群上的Kafka，并使用秘密管理器进行定期密码轮换。 问题： 我是否需要在每次密码轮换时重新实例化应用程序？如果是，有没有使用Spring Cloud Stream的标准方法？ 当前的解决方案:我们已经实现了一个基于Spring执行器的健康检查。在活页夹连接失败的情况下，我们重新启动ECS。 当前的解决方案非常不稳定，因为我们仍然会遇到SASLAuthentication异常，运行状况检查没有捕获到该异常，也没有重新实例化应用程序 请给我建议一条出路。
浏览 67提问于2021-11-08得票数 0
所以基本上我看到的是，我正在使用@KafkaListener消费消息，我故意不想承认它们，所以消费者忽略它们，并在主题中侦听新的消息。因此，如果应用程序重新启动，使用者只会从最后提交的消息开始。因此，我希望避免这种情况，我希望消费者在不重新启动的情况下考虑未提交的消息。下面是我正在使用的配置和@kafkalistener。
ReceiverConfig.java
package com.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serializ
浏览 3提问于2019-12-05得票数 1
1回答
我有两个弹簧引导应用程序。我试图映射到application.properties文件中，这样消费者就可以接收生产者发送的消息。我想补充一下，我使用的是一个CustomMessage：
public class CustomMessage {
private LocalDateTime timestamp;
private Integer sensor_id;
private Double measurement_value;
// getters and setters
}
我的生产者的application.properties文件：
# other propertie
浏览 2提问于2021-12-02得票数 2
回答已采纳
我已经用spring引导为我的kafka消费者编写了一个集成测试，与spring库一起使用。此测试使用EmbeddedKafka。使用一个分区的主题。为此，我使用了KafkaMessageListener容器。但我在这一行里有个错误
ContainerTestUtils.waitForAssignment(container, embeddedKafkaBroker.getPartitionsPerTopic())
我得到的错误是：
java.lang.IllegalStateException:预期的1，但有0分区。
我提到的代码是：
@EmbeddedKafka (partitions
浏览 6提问于2021-09-23得票数 0
回答已采纳
我查看了这些文档，发现如下：
spring.kafka.producer.properties.spring.json.type.mapping=cat:com.mycat.Cat,hat:com.myhat.Hat
我的制片人的财产：
spring.kafka.producer.bootstrap-server=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-seriali
浏览 4提问于2022-09-16得票数 0
回答已采纳
我有一个Kafka话题，有多个消费群体。我需要主题上的消息在它们的持久化持续时间到期时不被删除，如果它们还没有被所有消费者组阅读。
是否可以在持续时间之外设置额外的持久性规则？我需要消息始终停留在一个主题上，如果它们从未被消费过。
如果消息尚未被使用并且其持续时间已过期，是否可以“刷新”消息的超时时间？
浏览 0提问于2018-05-29得票数 2
我正在编写一个Kafka消费者，为了学习的目的，这一次我考虑使用Spring-Kafka实现。到目前为止，我一直使用Java API来编写使用者。 我想手动管理偏移量，所以我在Spring-Kafka包中搜索类似于ConsumerRebalanceListener的东西。我的成功之处在于，我在Spring中遇到了可以用来代替ConsumerRebalanceListener的ConsumerAwareRebalanceListener。 但是当我查看ConsumerAwareRebalanceListener接口时，可以看到两个方法- onPartitionsRevokedBeforeCom
浏览 49提问于2019-04-10得票数 0
1回答
卡夫卡流图()内如何等待未来？
、、、、
我正在Java中实现Spring应用程序，使用Spring和Kafka流绑定器。
我需要在KStream映射方法中实现阻塞操作，如下所示：
public Consumer<KStream<?, ?>> sink() {
return input -> input
.mapValues(value -> methodReturningCompletableFuture(value).get())
.foreach((key, value) -> otherMethod(key, value));
}
complet
浏览 3提问于2022-02-11得票数 -1
回答已采纳
我目前正在使用Spring来使用来自主题的消息以及@Retry of Spring。因此，基本上，我正在重新尝试处理消费者的信息，以防出现错误。但是，在这样做时，我希望避免KafkaMessageListenerContainer抛出的异常消息。相反，我想显示一条自定义消息。我尝试在ConcurrentKafkaListenerContainerFactory中添加一个错误处理程序，但是在这样做时，不会调用我的重试。有人能指导我如何显示自定义异常消息以及@Retry场景吗？下面是我的代码片段：
ConcurrentKafkaListenerContainerFactory Bean Confi
浏览 14提问于2020-06-24得票数 0
1回答
有没有办法在不重启spring kafka容器的情况下(并且在元数据最大寿命之前)手动触发重新平衡？从Kafka 2.6开始，就有了enforceRebalance()来触发来自消费者API的重新平衡，但它在spring kafka容器中是私有的。
浏览 26提问于2021-11-22得票数 0
回答已采纳
我正在尝试用Spring 2.1.7启动Spring，使用Java 12在本地主机上使用localhost。
获取错误："org.apache.kafka.clients.NetworkClient :clientId= Consumer 1，无法建立到节点-1的groupId=inter连接。代理可能不可用。“
我尝试将Java版本切换到11和8以及各种属性
spring:
kafka:
consumer:
#bootstrap-servers: localhost:9092
group-id: inter
auto-offset-re
浏览 1提问于2019-08-22得票数 9
是否可以在application.yml中仅为应用程序生成的主题启用spring.cloud.stream.kafka.streams.binder.auto-create-topics，而不是自动创建应用程序使用的主题？我希望我的制作人只负责创建主题。我正在使用spring-cloud-stream和kafka-streams绑定器。 在更一般的意义上，这是创建kafka主题的正确方法吗？
浏览 30提问于2020-11-24得票数 0
回答已采纳
1回答
我对Kakfa Spring集成非常陌生。我已经实现了卡夫卡的信息发送和一个听众，它的工作对我很好。但我想在两个地方听同样的信息。有人能帮我吗。下面是我的密码。
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.auto-commit-interval=100
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.gr
浏览 5提问于2017-03-02得票数 1
回答已采纳
我有一个spring boot应用程序(spring version 2.2.2.RELEASE)，我在其中配置了Kafka消费者，它处理来自Kafka的数据，并为多个web套接字提供服务。订阅kafka成功，但并不是所有选中Kafka topic的消息都会被消费者处理。很少有消息被延迟，也很少有消息被遗漏。但生产者正在发送数据，这是完全有保障的。下面我已经分享了我所使用的配置属性。 @Bean
public ConsumerFactory<String, String> consumerFactory() {
final String BOOTSTRAP_SERVERS
浏览 8提问于2020-01-10得票数 2
这就是场景:我知道，使用与Spring相关的最新API (比如kafka 2.10)，我们可以这样做：
@KafkaListener(id = "id0", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic", partitions = { "0" }) })
@KafkaListener(id = "id1", topicPartitions = { @TopicPartition(topic = "SpringKafkaTopic",
浏览 0提问于2017-04-17得票数 1
我刚刚意识到，在Kafka主题上收到的消息丢失了标题。有没有办法在发布到Kafka Topic之前给消息添加头部，然后在消费者端阅读？我使用的是Java11，Spring Cloud Hoxton.SR6，Spring2.2.4，kafka_2.13-2.6.0。提前感谢！
浏览 0提问于2020-09-08得票数 0
1回答
我使用spring 1.3.5，它使用0.11.0.2客户端。我有两个分区。我的max.poll.records = 1。
我的问题是:假设我有2个节点/工作人员(其并发性设置为一个，因此每个节点中有一个使用者)是慢速工作人员(处理记录可能需要5分钟)，这两个节点当前都处于关闭状态。当节点-1(使用者-1)加入消费者组时，它得到分配给它的两个分区(p-1，p-2)。现在，使用者-1消耗来自p-1的消息。节点2(消费者-2)立即加入相同的消费者组。
有没有可能P1被分配给消费者-2或者kafka知道消费者-1正在处理消息，所以它将分区-2分配给消费者-2？如果发生这种情况，节点2能否从p-1
浏览 1提问于2018-05-24得票数 1
我想知道是否可以使用spring-kafka和spring-kafka-test为@KafkaListener编写单元测试。
我的监听器类。
public class MyKafkaListener {
@Autowired
private MyMessageProcessor myMessageProcessor;
@KafkaListener(topics = "${kafka.topic.01}", groupId = "SF.CLIENT", clientIdPrefix = "SF.01", conta
浏览 3提问于2018-10-12得票数 11
我使用spring 2.3.2.RELEASE与spring -> 2.5.4 clients -> 2.5.0
我有以下简单的侦听器
@Slf4j
@Component
class SampleKafkaListener {
private final ObjectMapper objectMapper;
@Autowired
SampleKafkaListener(ObjectMapper objectMapper) {
this.objectMapper = objectMapper;
}
@KafkaListene
浏览 9提问于2020-08-25得票数 0
我正在编写卡夫卡消费者单元测试，并需要模拟我的KafkaConsumer的服务测试卡夫卡消费者独立。但是，服务的mockObject并没有被调用，而是由Spring创建并调用原始的服务类对象。因此，我的模拟类对象没有被调用。
KafkaConsumer：
@Slf4j
@Component
@RequiredArgsConstructor (onConstructor = @__(@Autowired))
public class KafkaEventConsumer {
private final MyService requestService;
@KafkaListener
浏览 87提问于2018-03-06得票数 0
回答已采纳
我们的应用程序被部署到一个k8s集群中，在该集群中，TLS证书每小时都会被非常积极地更新。应用程序还使用在application.yml中配置的相互TLS身份验证连接到Kafka。
据我所知，密钥库和信任库都以一种不可变的方式加载到SSLContext中，而且不能很容易地替换它。最简单、看似可行的解决方案可能是重新启动所有侦听器，如下代码所示：
@Autowired
private KafkaListenerEndpointRegistry messageListenerContainer;
@PostConstruct
public void setupTlsCertificateRefr
浏览 4提问于2021-12-17得票数 0
2回答
我正在尝试使用spring-boot和spring进行集成测试。我能够向spring嵌入式Kafka服务器生成消息，但是服务类中的侦听器尝试使用记录和
具有所有bean的KafkaProducerConfigTest Config类
@EnableKafka
@TestConfiguration
public class KafkaProducerConfigTest {
@Bean
public EmbeddedKafkaBroker embeddedKafkaBroker() {
return new EmbeddedKafkaBroker(1,false,2,"test
浏览 2提问于2019-02-19得票数 1
回答已采纳
我从卡夫卡开始。 我有一个包含两个代理(id #2和#3)的集群，复制因子为2。如果我断开一个代理(id #3)，我想测试Kafka的行为。 关闭#3后，我的主题信息： Topic: CUSTOMER PartitionCount: 5 ReplicationFactor: 2 Configs:
Topic: CUSTOMER Partition: 0 Leader: 2 Replicas: 3,2 Isr: 2
Topic: CUSTOMER Partition: 1 Leader: 2 Replic
浏览 122提问于2020-04-01得票数 1
回答已采纳
1回答
Camel JMS组件多线程
、、、、
我有一个spring boot应用程序，使用Camel的jmsComponent处理来自多个Solace队列的消息。我为每个队列连接定义了一个Camel route，并定义了一个将消息转换为json的Camel Processor。有大量来自队列的消息，而我的程序无法跟上速度。 路由： @Component
public class SomeRoute extends RouteBuilder {
@Override
public void configure() throws Exception {
from("jms:{{queue.name}}&
浏览 22提问于2020-07-21得票数 1
如果Kafka (spring.cloud.stream.bindings..producer.partitionCount)大于1并且Kafka (spring.cloud.stream.bindings..consumer.partitioned)为false，会发生什么情况(使用partitionCount )
浏览 28提问于2021-05-08得票数 0
回答已采纳
1回答
我有一个使用Spring实现的Kafka消费者：
public class MyKafkaConsumer<T> implements AcknowledgingMessageListener<String, T> {
private final ConcurrentMessageListenerContainer<String, T> container;
public MyKafkaConsumer(ConsumerFactory<String, T> consumerFactoryry) {
Container
浏览 0提问于2021-06-08得票数 0
我正在使用Spring-Kafka框架实现来自Kafka主题的消息消费。我正在尝试理解我为我的卡夫卡监听器创建的ConcurrentKafkaListenerContainerFactory的一些用法。@KafkaListener运行良好，但不出所料，在我的场景中，我有多个独立的侦听器，分别侦听多个主题。我想知道我是否可以在我所有的监听器中重用ConcurrentKafkaListenerContainerFactory，或者我必须为每个@KafkaListener创建一个containerFactory。有没有办法让所有@Kafkalisteners共享一个通用的containerFact
浏览 187提问于2020-04-26得票数 1
回答已采纳
我刚接触Kafka spring集成和spring云数据流。所以我的问题是，我正在尝试使用http is网关构建具有请求/回复交换的http post。
为了获得下面的流
:requestchannel > processor > replychannel
我当前的代码如下所示：
public class HttpSourceConfiguration {
@Autowired
private Source channels;
@Autowired
private HttpSourceProperties properties;
@Be
浏览 0提问于2019-03-05得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
腾讯会议活动推荐
腾讯云开发者
