Spring Boot与Reactor集成-Kafka的KafkaReceiver

使用Spring Boot与Reactor集成Kafka的KafkaReceiver

xml
    org.springframework.boot
    spring-boot-starter-webflux
    org.springframework.boot
    spring-boot-starter-kafka

java
@Configuration
public class KafkaReceiverConfig {
    
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;
    
    @Bean
    public KafkaReceiver kafkaReceiver() {
        ReceiverOptions receiverOptions = ReceiverOptions
                .create()
                .bootstrapServers(bootstrapServers)
                .groupId(groupId);
        
        return KafkaReceiver.create(receiverOptions);
    }
}

java
@Component
public class KafkaMessageReceiver {
    
    private final KafkaReceiver kafkaReceiver;
    
    public KafkaMessageReceiver(KafkaReceiver kafkaReceiver) {
        this.kafkaReceiver = kafkaReceiver;
    }
    
    public Flux> receiveMessages(String topic) {
        ReceiverOptions options = kafkaReceiver
                .receiverOptions()
                .subscription(Collections.singleton(topic));
        
        return kafkaReceiver.receive(options);
    }
}

java
@RestController
public class KafkaController {
    
    private final KafkaMessageReceiver kafkaMessageReceiver;
    
    public KafkaController(KafkaMessageReceiver kafkaMessageReceiver) {
        this.kafkaMessageReceiver = kafkaMessageReceiver;
    }
    
    @GetMapping("/messages")
    public Flux getMessages() {
        return kafkaMessageReceiver.receiveMessages("test-topic")
                .map(ReceiverRecord::value);
    }
}