diff --git a/src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java b/src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java
index 926f9d768..69338587d 100644
--- a/src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java
+++ b/src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java
@@ -26,12 +26,16 @@ import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationListener;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.data.keyvalue.core.AbstractKeyValueAdapter;
 import org.springframework.data.keyvalue.core.KeyValueAdapter;
@@ -99,17 +103,19 @@ import org.springframework.util.ObjectUtils;
  * @author Mark Paluch
  * @author Andrey Muchnik
  * @author John Blum
- * @author Lucian Torje
  * @since 1.7
  */
 public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
-		implements InitializingBean, ApplicationContextAware, ApplicationListener<RedisKeyspaceEvent> {
+		implements InitializingBean, SmartLifecycle, ApplicationContextAware, ApplicationListener<RedisKeyspaceEvent> {
 
 	/**
 	 * Time To Live in seconds that phantom keys should live longer than the actual key.
 	 */
 	private static final int PHANTOM_KEY_TTL = 300;
 
+	private final Log logger = LogFactory.getLog(getClass());
+	private final AtomicReference<State> state = new AtomicReference<>(State.CREATED);
+
 	private RedisOperations<?, ?> redisOps;
 	private RedisConverter converter;
 	private @Nullable RedisMessageListenerContainer messageListenerContainer;
@@ -121,6 +127,13 @@ public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
 	private @Nullable String keyspaceNotificationsConfigParameter = null;
 	private ShadowCopy shadowCopy = ShadowCopy.DEFAULT;
 
+	/**
+	 * Lifecycle state of this factory.
+	 */
+	enum State {
+		CREATED, STARTING, STARTED, STOPPING, STOPPED, DESTROYED;
+	}
+
 	/**
 	 * Creates new {@link RedisKeyValueAdapter} with default {@link RedisMappingContext} and default
 	 * {@link RedisCustomConversions}.
@@ -202,7 +215,7 @@ public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
 				&& this.expirationListener.get() == null) {
 
 			if (rdo.getTimeToLive() != null && rdo.getTimeToLive() > 0) {
-				initKeyExpirationListener();
+				initKeyExpirationListener(this.messageListenerContainer);
 			}
 		}
 
@@ -686,6 +699,11 @@ public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
 		this.shadowCopy = shadowCopy;
 	}
 
+	@Override
+	public boolean isRunning() {
+		return State.STARTED.equals(this.state.get());
+	}
+
 	/**
 	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
 	 * @since 1.8
@@ -696,22 +714,61 @@ public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
 		if (this.managedListenerContainer) {
 			initMessageListenerContainer();
 		}
+	}
 
-		if (ObjectUtils.nullSafeEquals(EnableKeyspaceEvents.ON_STARTUP, this.enableKeyspaceEvents)) {
-			initKeyExpirationListener();
+	@Override
+	public void start() {
+
+		State current = this.state.getAndUpdate(state -> isCreatedOrStopped(state) ? State.STARTING : state);
+
+		if (isCreatedOrStopped(current)) {
+
+			messageListenerContainer.start();
+
+			if (ObjectUtils.nullSafeEquals(EnableKeyspaceEvents.ON_STARTUP, this.enableKeyspaceEvents)) {
+				initKeyExpirationListener(this.messageListenerContainer);
+			}
+
+			this.state.set(State.STARTED);
+		}
+	}
+
+	private static boolean isCreatedOrStopped(@Nullable State state) {
+		return State.CREATED.equals(state) || State.STOPPED.equals(state);
+	}
+
+	@Override
+	public void stop() {
+
+		if (state.compareAndSet(State.STARTED, State.STOPPING)) {
+
+			KeyExpirationEventMessageListener listener = this.expirationListener.get();
+			if (listener != null) {
+
+				if (this.expirationListener.compareAndSet(listener, null)) {
+					try {
+						listener.destroy();
+					} catch (Exception e) {
+						logger.warn("Could not destroy KeyExpirationEventMessageListener", e);
+					}
+				}
+			}
+
+			messageListenerContainer.stop();
+			state.set(State.STOPPED);
 		}
 	}
 
 	public void destroy() throws Exception {
 
-		if (this.expirationListener.get() != null) {
-			this.expirationListener.get().destroy();
-		}
+		stop();
 
 		if (this.managedListenerContainer && this.messageListenerContainer != null) {
 			this.messageListenerContainer.destroy();
 			this.messageListenerContainer = null;
 		}
+
+		this.state.set(State.DESTROYED);
 	}
 
 	@Override
@@ -729,13 +786,12 @@ public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
 		this.messageListenerContainer = new RedisMessageListenerContainer();
 		this.messageListenerContainer.setConnectionFactory(((RedisTemplate<?, ?>) redisOps).getConnectionFactory());
 		this.messageListenerContainer.afterPropertiesSet();
-		this.messageListenerContainer.start();
 	}
 
-	private void initKeyExpirationListener() {
+	private void initKeyExpirationListener(RedisMessageListenerContainer messageListenerContainer) {
 
 		if (this.expirationListener.get() == null) {
-			MappingExpirationListener listener = new MappingExpirationListener(this.messageListenerContainer, this.redisOps,
+			MappingExpirationListener listener = new MappingExpirationListener(messageListenerContainer, this.redisOps,
 					this.converter, this.shadowCopy);
 
 			listener.setKeyspaceNotificationsConfigParameter(keyspaceNotificationsConfigParameter);
diff --git a/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java b/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java
index 1e23e01c6..65e57819f 100644
--- a/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java
+++ b/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java
@@ -82,19 +82,17 @@ public class RedisKeyValueAdapterTests {
 		adapter = new RedisKeyValueAdapter(template, mappingContext);
 		adapter.setEnableKeyspaceEvents(EnableKeyspaceEvents.ON_STARTUP);
 		adapter.afterPropertiesSet();
+		adapter.start();
 
 		template.execute((RedisCallback<Void>) connection -> {
 			connection.flushDb();
 			return null;
 		});
 
-		RedisConnection connection = template.getConnectionFactory().getConnection();
-
-		try {
+		try (RedisConnection connection = template.getConnectionFactory()
+				.getConnection()) {
 			connection.setConfig("notify-keyspace-events", "");
 			connection.setConfig("notify-keyspace-events", "KEA");
-		} finally {
-			connection.close();
 		}
 	}
 
diff --git a/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterUnitTests.java b/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterUnitTests.java
index b992cb5fa..dca24e748 100644
--- a/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterUnitTests.java
@@ -100,6 +100,7 @@ class RedisKeyValueAdapterUnitTests {
 
 		adapter = new RedisKeyValueAdapter(template, context);
 		adapter.afterPropertiesSet();
+		adapter.start();
 	}
 
 	@AfterEach
@@ -153,12 +154,32 @@ class RedisKeyValueAdapterUnitTests {
 		adapter = new RedisKeyValueAdapter(template, context);
 		adapter.setEnableKeyspaceEvents(EnableKeyspaceEvents.ON_STARTUP);
 		adapter.afterPropertiesSet();
+		adapter.start();
 
 		KeyExpirationEventMessageListener listener = ((AtomicReference<KeyExpirationEventMessageListener>) getField(adapter,
 				"expirationListener")).get();
 		assertThat(listener).isNotNull();
 	}
 
+	@Test // GH-2957
+	void adapterShouldBeRestartable() throws Exception {
+
+		adapter.destroy();
+
+		adapter = new RedisKeyValueAdapter(template, context);
+		adapter.setEnableKeyspaceEvents(EnableKeyspaceEvents.ON_STARTUP);
+		adapter.afterPropertiesSet();
+		adapter.start();
+		adapter.stop();
+
+		assertThat(((AtomicReference<KeyExpirationEventMessageListener>) getField(adapter, "expirationListener")).get())
+				.isNull();
+
+		adapter.start();
+		assertThat(((AtomicReference<KeyExpirationEventMessageListener>) getField(adapter, "expirationListener")).get())
+				.isNotNull();
+	}
+
 	@Test // DATAREDIS-491
 	void shouldInitKeyExpirationListenerOnFirstPutWithTtl() throws Exception {