Spring kafka @KafkaListener 未被调用答案

【问题标题】：Spring kafka @KafkaListener is not being invokedSpring kafka @KafkaListener 未被调用
【发布时间】：2019-09-16 20:12:45
【问题描述】：

我正在尝试使用 spring-kafka 将 kafka 与我的 Spring boot (v2.0.6.RELEASE) 应用程序集成。现在我想要一个消费者和一个生产者。我让生产者工作得很好，我可以看到通过控制台消费者发送到主题的消息。我无法让消费者代码工作，当新消息出现在 kafka 主题中时，它不会被调用。

这是我的kafka配置类：


@Configuration
@EnableKafka
public class KafkaConfiguration {

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();

        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, <path to my custom serializer>);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        return props;
    }

    @Bean
    public Map<String, Object> consumerConfigs(){
        Map<String, Object> props = new HashMap<>();

        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, <path to my custom deserializer>);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
//        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        return props;
    }

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

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, List<MyMessage>>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, List<MyMessage>> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

    @Bean
    public ProducerFactory<String, List<MyMessage>> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, List<MyMessage>> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}

这是我的 pom 依赖：

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

以及消费者代码：


@Slf4j
public class MyMessageConsumer {

    @KafkaListener(topicPattern = "test_topic")
    public void receive(List<MyMessage> myMessages){
        log.info("Received payload {}", myMessages);
    }
}

我正在我的计算机上运行 kafka，正如我所说 - 我可以看到控制台使用者中出现的消息。我的基于 Spring 的消费者收不到消息的原因可能是什么？

编辑：这是反序列化器类：


public class TagEventDeserializer implements Deserializer<List<MyMessage>> {
    @Override
    public void configure(Map map, boolean b) {

    }

    @Override
    public List<MyMessage> deserialize(String s, byte[] bytes) {
        ObjectMapper objectMapper = new ObjectMapper();
        List<MyMessage> myMessages = null;
        try{
            myMessages = objectMapper.readValue(bytes , List.class);
        } catch (Exception e){
            e.printStackTrace();
        }
        return myMessages;
    }

    @Override
    public void close() {

    }
}

【问题讨论】：

启用此props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 属性并更改消费者组重新启动应用程序
不幸的是，这没有帮助
你确定你的反序列化器在工作吗？您是否在应用程序中设置了日志记录以查看底层 Kafka 类中是否发生任何错误？
我认为这不是反序列化器的问题，因为反序列化器中的代码根本没有被调用。但以防万一我将代码添加到原始问题中

标签： java spring apache-kafka spring-kafka

【解决方案1】：

由于缺少构造型注释，spring 不会扫描您的类MyMessageConsumer，因此例如向此类添加@Component 将解决您的问题：

@Slf4j
@Component
public class MyMessageConsumer {

    @KafkaListener(topicPattern = "test_topic")
    public void receive(List<MyMessage> myMessages){
        log.info("Received payload {}", myMessages);
    }
}

我认为在 @KafkaListner 注释中指定一个消费者组是个好主意

【讨论】：