如何将@KafkaListener注释方法与Flux或Mono一起使用？

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

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

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

@Service
public class KafkaConsumerService {

    @KafkaListener(topics = "myTopic", groupId = "myGroupId")
    public Flux<String> consumeMessages(ConsumerRecord<String, String> record) {
        return Flux.fromIterable(Collections.singletonList(record.value()));
    }
}

@RestController
public class KafkaController {

    private final KafkaConsumerService consumerService;

    public KafkaController(KafkaConsumerService consumerService) {
        this.consumerService = consumerService;
    }

    @GetMapping("/messages")
    public Flux<String> getMessages() {
        return consumerService.consumeMessages();
    }
}