0

我正在使用 spring-kafka，并且正在尝试创建一个 Kafka 消费者。我参考了http://howtoprogram.xyz/2016/09/23/spring-kafka-tutorial/ 和https://www.codenotfound.com/2016/09/spring-kafka-consumer-producer-example.html . 我目前使用的是参考文献中完全相同的代码。

这是接收器类

public class Receiver {

private static final Logger LOGGER = LoggerFactory.getLogger(Receiver.class);
public final CountDownLatch countDownLatch1 = new CountDownLatch(1);

@KafkaListener(id = "foo", topics = "HelloKafkaTopic", group = "group1")
public void receive(ConsumerRecord<?, ?> record) {
    System.out.println(record);
    countDownLatch1.countDown();
 }
}

这是 KafkaConsumerConfig

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(3);
    factory.getContainerProperties().setPollTimeout(3000);
    return factory;
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}

@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> propsMap = new HashMap<>();
    propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
    propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
    propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
    propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    return propsMap;
}

@Bean
public Receiver listener() {
    return new Receiver();
}

只要我用@EnableKafka 注释我的类，它有@KafkaListener 方法。我收到以下错误。我很难理解我是否在某个地方犯了错误。

java.lang.NoSuchMethodError: org.springframework.messaging.handler.annotation.support.MessageMethodArgumentResolver.<init>(Lorg/springframework/messaging/converter/MessageConverter;)V
at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor$KafkaHandlerMethodFactoryAdapter.createDefaultMessageHandlerMethodFactory(KafkaListenerAnnotationBeanPostProcessor.java:654)
at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor$KafkaHandlerMethodFactoryAdapter.getMessageHandlerMethodFactory(KafkaListenerAnnotationBeanPostProcessor.java:630)
at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor$KafkaHandlerMethodFactoryAdapter.createInvocableHandlerMethod(KafkaListenerAnnotationBeanPostProcessor.java:625)
at org.springframework.kafka.config.MethodKafkaListenerEndpoint.configureListenerAdapter(MethodKafkaListenerEndpoint.java:112)
at org.springframework.kafka.config.MethodKafkaListenerEndpoint.createMessageListener(MethodKafkaListenerEndpoint.java:101)
at org.springframework.kafka.config.AbstractKafkaListenerEndpoint.setupMessageListener(AbstractKafkaListenerEndpoint.java:297)
at org.springframework.kafka.config.AbstractKafkaListenerEndpoint.setupListenerContainer(AbstractKafkaListenerEndpoint.java:282)
at org.springframework.kafka.config.AbstractKafkaListenerContainerFactory.createListenerContainer(AbstractKafkaListenerContainerFactory.java:211)
at org.springframework.kafka.config.AbstractKafkaListenerContainerFactory.createListenerContainer(AbstractKafkaListenerContainerFactory.java:46)
at org.springframework.kafka.config.KafkaListenerEndpointRegistry.createListenerContainer(KafkaListenerEndpointRegistry.java:182)
at org.springframework.kafka.config.KafkaListenerEndpointRegistry.registerListenerContainer(KafkaListenerEndpointRegistry.java:154)
at org.springframework.kafka.config.KafkaListenerEndpointRegistry.registerListenerContainer(KafkaListenerEndpointRegistry.java:128)
at org.springframework.kafka.config.KafkaListenerEndpointRegistrar.registerAllEndpoints(KafkaListenerEndpointRegistrar.java:138)
at org.springframework.kafka.config.KafkaListenerEndpointRegistrar.afterPropertiesSet(KafkaListenerEndpointRegistrar.java:132)
at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(KafkaListenerAnnotationBeanPostProcessor.java:224)
at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:792)
at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:839)
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:538)
at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:118)
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:766)
at org.springframework.boot.SpringApplication.createAndRefreshContext(SpringApplication.java:361)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:307)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1191)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1180)
4

1 回答 1

2

我认为问题出在您的类路径中，您使用的是旧版本的spring-messaging. spring-messaging修复使用依赖管理的最新（v4.3.x）版本pom.xml并尝试运行。

首先检查spring-messaging您当前使用的是哪个版本？

于 2017-04-19T05:26:45.627 回答