spring kafka：如果我在应用程序中定义了一个记录制作人，kafka消费者就不会收到记录

Question

我正在使用Spring和Spring Kafka编写一个小型PoC。我的目标是让生产者和消费者都从这个主题中写作（例如阅读）。

我有一个奇怪的情况：

• 制作人正确地制作了这些记录（我可以通过Python脚本使用它们）

下面是我的代码——它与文档示例非常相似。更准确地说，问题来自这样一个事实，即KafkanConsumerConfiguration中的bean不是由Spring创建的（也就是说，构建它们的方法永远不会被调用）。

Kafkaproducer配置。JAVA

@Configuration
public class KafkaProducerConfiguration {

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

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

消息ender.java

@Component
public class MessageSender {
    final static private Logger log = Logger.getLogger(MessageSender.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostConstruct
    public void onConstruct() throws InterruptedException {
        log.info("Sending messages...");
        for (int i = 0; i < 100; ++i) {
            kafkaTemplate.send("mytopic", "this is a message");
            Thread.sleep(1000);
        }
        kafkaTemplate.flush(); // NOTE: no changes if I move this call in the loop
        log.info("Done sending messages");
    }
}

Kafka消费者配置。JAVA

@Configuration
@EnableKafka
public class KafkaConsumerConfiguration {

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

        return factory;
    }

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

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:32768");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-service");
        return props;
    }
}

MyMessageListener。JAVA

@Service
public class MyMessageListener {
    final static private Logger log = Logger.getLogger(MyMessageListener.class);

    @PostConstruct
    public void onConstruct() {
        log.info("Message listener started");
    }

    @KafkaListener(topics = "mytopic")
    public void onMessageReceived(String message) {
        log.info("Got message: "+ message);
    }
}

以下是应用程序生成的日志，供参考：https://pastebin.com/BY783jiL.如您所见，没有创建消费bean（否则将出现一个块ConsumerConfig values:…）。

以下是我尝试过但没有成功的几件事：

• 将生产者和消费者配置放在同一个配置类中
• 更改Kafka消费者配置中bean的名称（并在MyMessageListener.onMessageReception方法上添加注释属性容器工厂="myBeanName"）
• 将类的名称Kafka消费者配置更改为其他内容
• 在我的Kafka消费者配置中添加一个与kafka无关的@Bean以查看它是否会被创建（它会）

版本：Spring Boot 1.5.9，Spring-Kafka：1.1.7。

我已经撕了几个小时的头发了，非常感谢您的帮助。

谢谢

Answer 1

刚刚发现问题<代码>消息发送者。onConstruct实际上需要花费大量时间来执行（100秒），在此期间，它会阻止Spring创建其他bean。

Answer 2

kafkaTemplate.send("mytopic", "this is a message");

您永远不应该以@PostConstruct方法开始与外部服务交互——您需要等待应用程序构建完成后再这样做。

实现SmartLifeyle，为isAutoStartup返回true并将该代码移动到start（）。

或者实现ApplicationListener

无论哪种方式都将确保应用程序已准备就绪。