如何在 Spring MVC Web 应用程序中实现 Kafka 消费者（使用 Spring Boot）

Question

我想在 Spring MVC Web 应用程序中创建一个使用者。基本上，我希望网络应用程序可以收听 Kafka 上的一些主题，并根据收到的消息采取一些措施。

到目前为止，我所看到的所有示例都使用具有无限循环的独立应用程序（使用纯 java）或在单元测试中（使用 spring）：

@Autowired
private Listener listener;

@Autowired
private KafkaTemplate<Integer, String> template;

@Test
public void testSimple() throws Exception {
    template.send("annotated1", 0, "foo");
    template.flush();
    assertTrue(this.listener.latch1.await(10, TimeUnit.SECONDS));
}

@Configuration
@EnableKafka
public class Config {

@Bean
ConcurrentKafkaListenerContainerFactory<Integer, String>
                    kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
}

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

@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
    ...
    return props;
}

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

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

@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
    ...
    return props;
}

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

}
public class Listener {

    private final CountDownLatch latch1 = new CountDownLatch(1);

    @KafkaListener(id = "foo", topics = "annotated1")
        public void listen1(String foo) {
            this.latch1.countDown();
     }

}

创建侦听器的最佳位置是什么？我应该把它放在那里：

@SpringBootApplication
public class Application {

@Autowired
private Listener listener;

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

}