在尝试自动连接时在spring托管类上获取空指针

Question

我正在创建一个spring引导应用程序来从Rabbit队列中侦听消息并将消息转发到Kafka流。我试图@Autowire一个Kafka配置类，但我得到NullPointerException，更令人惊讶的是，当我列出所有的spring管理bean，然后它给了我的类的对象名称，我试图@Autowire。我不知道我错过了什么。

开始类:

@SpringBootApplication
@ComponentScan(basePackages = {"com.example.demo.service","com.example.demo.configuration"})
public class RabbitConsumerApplication {

private static ApplicationContext applicationContext;
public static void main(String[] args) {
applicationContext = SpringApplication.run(RabbitConsumerApplication.class, args);
displayAllBeans();
}

public static void displayAllBeans() {
String[] allBeanNames = applicationContext.getBeanDefinitionNames();
for(String beanName : allBeanNames) {
System.out.println(beanName);
}
}

}

KafkaProducerService类:

package com.example.demo.service;
@Component
public class KafkaProducerService {


@Autowired
private KafkaTemplate<?, String> kafkaTemplate;


public void getMessageFromMQ(Message  message) throws InterruptedException {

Map<String, Object> configProps = new HashMap<>();
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
DefaultKafkaProducerFactory<Object, Object> defaultKafkaProducerFactory = new DefaultKafkaProducerFactory<>(configProps);


kafkaTemplate.send("testTopic",new String(message.getBody()));

}

@Override
public String toString() {
return "KafkaProducerService [kafkaTemplate=" + kafkaTemplate + "]";
}


}

KafkaProducerConfig类:

package com.example.demo.configuration;
@Configuration
public class KafkaProducerConfig {

@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}

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

RabbitListenerService类:

package com.example.demo.service;
public class RabbitListenerService implements MessageListener{

@Autowired
private KafkaProducerService kafkaProducerService;


@Override
public void onMessage(Message message) {
System.out.println("message :"+new String(message.getBody()));
System.out.println("object:"+kafkaProducerService.toString());  //19th line. getting nullpointer on this line   
}
}

堆栈跟踪:

2021-02-20 13:55:06.750  WARN 5392 --- [tnerContainer-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener threw exception
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:1746) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1692) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1562) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1539) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1530) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1474) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:967) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:913) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1288) ~[spring-rabbit-2.3.4.jar:2.3.4]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1194) ~[spring-rabbit-2.3.4.jar:2.3.4]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
Caused by: java.lang.NullPointerException: null
at com.example.demo.service.RabbitListenerService.onMessage(RabbitListenerService.java:19) ~[classes/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1688) ~[spring-rabbit-2.3.4.jar:2.3.4]
... 10 common frames omitted
message :hey you
message :hey you

我只是想把@AutowireKafkaProducerService类变成RabbitListenerService。

Answer 1

好了，我算是解决了这个问题。即使@autowire正在发生，但它实际上绑定了一个空对象，因为在RabbitConfig.java中，我实际上用一个新关键字初始化了rabbitListenerService，并且spring IOC不处理。我改变了下面的方法，它解决了问题。

RabbitListenerService.java

@Component
public class RabbitListenerService implements MessageListener{

@Autowired
private KafkaProducerService kafkaProducerService;


@Override
public void onMessage(Message message) {
System.out.println("message :"+new String(message.getBody()));
System.out.println("object:"+kafkaProducerService.toString());

try {
kafkaProducerService.getMessageFromMQ(message);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

然后自动将RabbitListenerService连接到RabbitConfig

RabbitConfig.java

@Configuration
public class RabbitConfig {

private static final String my_queue = "TestQueue";

@Autowired
private RabbitListenerService rabbitListenerService;


@Bean
MessageListenerContainer messageListnerContainer() {

SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer();
simpleMessageListenerContainer.setConnectionFactory(connectionFactory());
simpleMessageListenerContainer.setQueues(myQueue());
simpleMessageListenerContainer.setMessageListener(rabbitListenerService);
return simpleMessageListenerContainer;
}

}

Answer 2

从你的代码来看，你似乎没有定义任何kafka模板bean。例如:

@Configuration
public class MyKafkaConfig {
...
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
// define a template here
}
}

kafka服务定义了一个注入(自动连接)的KafkaTemplate<?, String>。在配置类中定义了这样一个bean吗?

也请张贴你的stacktrace。

Answer 3

试试这个，用*替换包，确保它在bean列表中打印KafkaProducerService，

始终实践为impl服务类创建接口

@Component
public class KafkaProducerServiceImpl implements KafkaProducerService

。

@SpringBootApplication
@ComponentScan(basePackages = "com.example.demo.*"})
public class RabbitConsumerApplication {

...
}