diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarConsumerFactory.java
new file mode 100644
index 00000000..fe57d466
--- /dev/null
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarConsumerFactory.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.core.reactive;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.reactive.client.api.ImmutableReactiveMessageConsumerSpec;
+import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerSpec;
+import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
+
+import org.springframework.util.CollectionUtils;
+
+/**
+ * Default implementation for {@link ReactivePulsarConsumerFactory}.
+ *
+ * @param <T> underlying payload type for the reactive consumer.
+ * @author Christophe Bornet
+ */
+public class DefaultReactivePulsarConsumerFactory<T> implements ReactivePulsarConsumerFactory<T> {
+
+	private final ReactiveMessageConsumerSpec consumerSpec;
+
+	private final ReactivePulsarClient reactivePulsarClient;
+
+	public DefaultReactivePulsarConsumerFactory(ReactivePulsarClient reactivePulsarClient,
+			ReactiveMessageConsumerSpec consumerSpec) {
+		this.consumerSpec = new ImmutableReactiveMessageConsumerSpec(
+				consumerSpec != null ? consumerSpec : new MutableReactiveMessageConsumerSpec());
+		this.reactivePulsarClient = reactivePulsarClient;
+	}
+
+	@Override
+	public ReactiveMessageConsumer<T> createConsumer(Schema<T> schema) {
+		return createConsumer(schema, Collections.emptyList());
+	}
+
+	@Override
+	public ReactiveMessageConsumer<T> createConsumer(Schema<T> schema,
+			List<ReactiveMessageConsumerBuilderCustomizer<T>> customizers) {
+
+		ReactiveMessageConsumerBuilder<T> consumer = this.reactivePulsarClient.messageConsumer(schema);
+
+		consumer.applySpec(this.consumerSpec);
+		if (!CollectionUtils.isEmpty(customizers)) {
+			customizers.forEach((c) -> c.customize(consumer));
+		}
+
+		return consumer.build();
+	}
+
+}
diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageConsumerBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageConsumerBuilderCustomizer.java
new file mode 100644
index 00000000..c83f0b22
--- /dev/null
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageConsumerBuilderCustomizer.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.core.reactive;
+
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
+
+/**
+ * The interface to customize a {@link ReactiveMessageConsumerBuilder}.
+ *
+ * @param <T> The message payload type
+ * @author Christophe Bornet
+ */
+@FunctionalInterface
+public interface ReactiveMessageConsumerBuilderCustomizer<T> {
+
+	/**
+	 * Customizes a {@link ReactiveMessageConsumerBuilder}.
+	 * @param reactiveMessageConsumerBuilder the builder to customize
+	 */
+	void customize(ReactiveMessageConsumerBuilder<T> reactiveMessageConsumerBuilder);
+
+}
diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarConsumerFactory.java
new file mode 100644
index 00000000..fc7870af
--- /dev/null
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarConsumerFactory.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.core.reactive;
+
+import java.util.List;
+
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
+
+/**
+ * Pulsar reactive consumer factory interface.
+ *
+ * @param <T> payload type for the consumer.
+ * @author Christophe Bornet
+ */
+public interface ReactivePulsarConsumerFactory<T> {
+
+	/**
+	 * Create a reactive message consumer.
+	 * @param schema the schema of the messages to be consumed
+	 * @return the reactive message consumer
+	 */
+	ReactiveMessageConsumer<T> createConsumer(Schema<T> schema);
+
+	/**
+	 * Create a reactive message consumer.
+	 * @param schema the schema of the messages to be consumed
+	 * @param customizers the optional list of customizers to apply to the reactive
+	 * message consumer builder
+	 * @return the reactive message consumer
+	 */
+	ReactiveMessageConsumer<T> createConsumer(Schema<T> schema,
+			List<ReactiveMessageConsumerBuilderCustomizer<T>> customizers);
+
+}
diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageConsumerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageConsumerFactoryTests.java
new file mode 100644
index 00000000..7e426783
--- /dev/null
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageConsumerFactoryTests.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.core.reactive;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
+import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerSpec;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link DefaultReactivePulsarConsumerFactory}
+ *
+ * @author Christophe Bornet
+ */
+class DefaultReactiveMessageConsumerFactoryTests {
+
+	private static final Schema<String> SCHEMA = Schema.STRING;
+
+	@Test
+	void createConsumer() {
+		MutableReactiveMessageConsumerSpec spec = new MutableReactiveMessageConsumerSpec();
+		spec.setConsumerName("test-consumer");
+		DefaultReactivePulsarConsumerFactory<String> consumerFactory = new DefaultReactivePulsarConsumerFactory<>(
+				AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec);
+
+		ReactiveMessageConsumer<String> consumer = consumerFactory.createConsumer(SCHEMA);
+
+		assertThat(consumer)
+				.extracting("consumerSpec", InstanceOfAssertFactories.type(ReactiveMessageConsumerSpec.class))
+				.extracting(ReactiveMessageConsumerSpec::getConsumerName).isEqualTo("test-consumer");
+	}
+
+	@Test
+	void createConsumerWithCustomizer() {
+		MutableReactiveMessageConsumerSpec spec = new MutableReactiveMessageConsumerSpec();
+		spec.setConsumerName("test-consumer");
+		DefaultReactivePulsarConsumerFactory<String> consumerFactory = new DefaultReactivePulsarConsumerFactory<>(
+				AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec);
+
+		ReactiveMessageConsumer<String> consumer = consumerFactory.createConsumer(SCHEMA,
+				Collections.singletonList(builder -> builder.consumerName("new-test-consumer")));
+
+		assertThat(consumer)
+				.extracting("consumerSpec", InstanceOfAssertFactories.type(ReactiveMessageConsumerSpec.class))
+				.extracting(ReactiveMessageConsumerSpec::getConsumerName).isEqualTo("new-test-consumer");
+	}
+
+}