Reactor/Reactive Kafka 中有 @KafkaListener 吗？

Question

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

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

它在运行和停止。我希望它总是等待新消息。

Answer 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 - 启动zookeeper（默认设置） 2 - 启动kafka服务器（默认设置） 3 - 创建一个名为test的主题 4 - 生成一条简单的消息 5 - 在spring初始化程序中创建一个简单的项目并添加reactor-kafka依赖并显然相应地添加上面的服务

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