通过 spring kafka 将控制器休息到 return kafka 中的记录

Question

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

@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();
    }

}

和另一个 class 如下消费者

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()。

通过 spring kafka 将控制器休息到 return kafka 中的记录

rest controller to return records in kafka via spring kafka

spring-boot

apache-kafka

kafka-consumer-api

spring-kafka

spring-web