下面是一个例子：

// Import the necessary classes
import org.apache.kafka.common.serialization.StringDeserializer;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.receiver.KafkaReceiver;

public class ReactiveKafkaExample {
    public static void main(String[] args) {
        // Create a receiver options object to configure the Kafka receiver
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(consumerProps)
                .subscription(Collections.singleton("my-topic"))
                .addAssignListener(partitions -> System.out.println("onPartitionsAssigned {}" + partitions))
                .addRevokeListener(partitions -> System.out.println("onPartitionsRevoked {}" + partitions));

        // Create a Kafka receiver using the receiver options
        KafkaReceiver<String, String> kafkaReceiver = KafkaReceiver.create(receiverOptions);

        // Use the `flatMap` operator to apply back-pressure to the receiver
        kafkaReceiver.flatMap(message -> {
            // Handle the incoming message
            // ...
            return Mono.empty();
        });
    }
}