休息控制器通过spring kafka返回kafka中的记录

Question

对于我的演示应用程序，我必须创建一个休息控制器来返回 kafka 队列中的消息。我已阅读 spring-kafka 参考指南并实现了消费者配置并创建了如下 bean

@Configuration
@EnableKafka
public class ConsumerConfiguration {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        // list of host:port pairs used for establishing the initial connections to the Kakfa cluster
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        // allows a pool of processes to divide the work of consuming and processing records
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "trx");

        return props;
    }

    @Bean
    public ConsumerFactory<String, Transaction> transactionConsumerFactory() {
        return new DefaultKafkaConsumerFactory<>(
                consumerConfigs(),new StringDeserializer(),new JsonDeserializer<>(Transaction.class)
        );
    }

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

        return factory;
    }

    @Bean
    public Consumer consumer() {
        return new Consumer();
    }

}

以及像下面这样的另一个类消费者

public class Consumer {

    private CountDownLatch latch = new CountDownLatch(1);

    public CountDownLatch getLatch() {
        return latch;
    }

    @KafkaListener(topics = "${kafka.topic.name}")
    public void receive(Transaction transaction) {
        latch.countDown();
    }
}

我现在如何实现逻辑，以便在每次在控制器上命中时从消费者那里检索事务。

提前致谢。

Answer 1

好吧，@KafkaListener 产生独立的长期进程，将记录从 Kafka 流式传输到回调。由于您正在谈论 REST GET 事件，因此您别无选择，除非从 ConsumerFactory 获取 KafkaConsumer 并从控制器方法手动调用其 poll()。