我正在尝试使用spring-kafka将kafka与我的Spring启动(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() {
}
}
因为你的类MyMessageConsumer
由于缺少刻板印象注释而没有被春天扫描,所以在这个类中添加一个@Component
将解决你的问题:
@Slf4j
@Component
public class MyMessageConsumer {
@KafkaListener(topicPattern = "test_topic")
public void receive(List<MyMessage> myMessages){
log.info("Received payload {}", myMessages);
}
}
我认为在@KafkaListner
注释中指定一个消费者群体是个好主意