Reactor/Reactive Kafka 中有 @KafkaListener 吗？答案

【问题标题】：Is there @KafkaListener in Reactor/Reactive Kafka?Reactor/Reactive Kafka 中有 @KafkaListener 吗？
【发布时间】：2019-04-09 03:09:39
【问题描述】：

我想写 Reactive Kafa 来收听新的消息。但我不知道该怎么做。就像 @KafkaListener 阻止 Kafka - 它正在等待新消息

Spring Boot Webflux 和 Reactor Kafka 上的代码演示：

public class KafkaConsumer {
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TOPIC = "Kafka_Example";

    private final ReceiverOptions<String, String> receiverOptions;

    public KafkaConsumer(String bootstrapServers) {

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "sample-consumer");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "sample-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        receiverOptions = ReceiverOptions.create(props);

    }

    public Disposable consumeMessages(String topic) {

        ReceiverOptions<String, String> options = receiverOptions.subscription(Collections.singleton(topic))
                .addAssignListener(partitions -> System.out.println("onPartitionsAssigned " + partitions))
                .addRevokeListener(partitions -> System.out.println("onPartitionsRevoked " + partitions));

        Flux<ReceiverRecord<String, String>> kafkaFlux = KafkaReceiver.create(options).receive();
        return kafkaFlux.subscribe(record -> {
            ReceiverOffset offset = record.receiverOffset();
            System.out.printf("Received message: offset=%d key=%d value=%s\n", offset.offset(), record.key(), record.value());
            offset.acknowledge();
        });
    }

    public static void main(String[] args) throws Exception {
        KafkaConsumer consumer = new KafkaConsumer(BOOTSTRAP_SERVERS);
        consumer.consumeMessages(TOPIC);
    }
}

它正在运行和停止。我希望它一直在等待新消息。

【问题讨论】：

您找到解决方案了吗？

标签： spring spring-boot apache-kafka reactive

【解决方案1】：

package com.simplest.kafkaconsumer.services;



import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import reactor.core.publisher.Flux;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.receiver.ReceiverRecord;

@Component
public class ReactorKafkaReceiver {

    private static final Logger log = LoggerFactory.getLogger(ReactorKafkaReceiver.class.getName());

    private KafkaReceiver kafkaReceiver;

    public ReactorKafkaReceiver() {


        final Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "test");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        ReceiverOptions<Object, Object> consumerOptions = ReceiverOptions.create(consumerProps)
                .subscription(Collections.singleton("test"))
                .addAssignListener(partitions -> log.debug("onPartitionsAssigned {}", partitions))
                .addRevokeListener(partitions -> log.debug("onPartitionsRevoked {}", partitions));

        kafkaReceiver = KafkaReceiver.create(consumerOptions);

        ((Flux<ReceiverRecord>) kafkaReceiver.receive())
                .doOnNext(r -> {
                    System.out.println(r.value());
                    r.receiverOffset().acknowledge();
                })
                .subscribe();
    }

}

如果你想测试： 1 - startzookeeper（默认设置） 2 - 启动 kafka 服务器（默认设置） 3 - 创建一个名为 test 的主题 4 - 产生一个简单的消息 5 - 在 spring 初始化器中创建一个简单的项目并添加 reactor-kafka 依赖，并相应地添加上面的服务

我想知道我是否应该粘贴作为答案，因为您的主题问题是：Reactor/Reactive Kafka 中是否有 @KafkaListener？但是在阅读了两次您的问题之后，您似乎主要面临继续聆听的问题（它正在运行和停止。我希望它一直在等待新消息。）

【讨论】：

【解决方案2】：

我的应用程序也需要相同的内容。目前我在 GitHub https://github.com/reactor/reactor-kafka/issues/100 上发现了这个问题（有人提出了解决方案，我还没测试）。

https://github.com/spring-projects/spring-kafka/pull/1123 提到了未来的 Kafka Reactor 更改，但最近此 Pull Request 已关闭。

【讨论】：