diff --git a/src/main/java/org/springframework/data/redis/core/AbstractOperations.java b/src/main/java/org/springframework/data/redis/core/AbstractOperations.java
index d3f2a080d..3d0832e93 100644
--- a/src/main/java/org/springframework/data/redis/core/AbstractOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/AbstractOperations.java
@@ -23,6 +23,7 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.BiFunction;
 
 import org.springframework.data.geo.GeoResults;
 import org.springframework.data.redis.connection.RedisConnection;
@@ -51,7 +52,7 @@ abstract class AbstractOperations<K, V> {
 
 	// utility methods for the template internal methods
 	abstract class ValueDeserializingRedisCallback implements RedisCallback<V> {
-		private Object key;
+		private final Object key;
 
 		public ValueDeserializingRedisCallback(Object key) {
 			this.key = key;
@@ -66,12 +67,31 @@ abstract class AbstractOperations<K, V> {
 		protected abstract byte[] inRedis(byte[] rawKey, RedisConnection connection);
 	}
 
+	private class FunctionalValueDeserializingRedisCallback extends ValueDeserializingRedisCallback {
+
+		private final BiFunction<RedisConnection, byte[], byte[]> function;
+
+		public FunctionalValueDeserializingRedisCallback(Object key, BiFunction<RedisConnection, byte[], byte[]> function) {
+			super(key);
+			this.function = function;
+		}
+
+		@Nullable
+		protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+			return function.apply(connection, rawKey);
+		}
+	}
+
 	final RedisTemplate<K, V> template;
 
 	AbstractOperations(RedisTemplate<K, V> template) {
 		this.template = template;
 	}
 
+	ValueDeserializingRedisCallback valueCallbackFor(Object key, BiFunction<RedisConnection, byte[], byte[]> function) {
+		return new FunctionalValueDeserializingRedisCallback(key, function);
+	}
+
 	RedisSerializer keySerializer() {
 		return template.getKeySerializer();
 	}
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java
index c3d3e00e5..6b857ee59 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java
@@ -22,8 +22,9 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+
 import org.springframework.data.redis.connection.BitFieldSubCommands;
-import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.DefaultedRedisConnection;
 import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
 import org.springframework.data.redis.core.types.Expiration;
 import org.springframework.lang.Nullable;
@@ -45,79 +46,39 @@ class DefaultValueOperations<K, V> extends AbstractOperations<K, V> implements V
 
 	@Override
 	public V get(Object key) {
-
-		return execute(new ValueDeserializingRedisCallback(key) {
-
-			@Override
-			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
-				return connection.get(rawKey);
-			}
-		});
+		return execute(valueCallbackFor(key, DefaultedRedisConnection::get));
 	}
 
 	@Nullable
 	@Override
 	public V getAndDelete(K key) {
-
-		return execute(new ValueDeserializingRedisCallback(key) {
-
-			@Override
-			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
-				return connection.getDel(rawKey);
-			}
-		});
+		return execute(valueCallbackFor(key, DefaultedRedisConnection::getDel));
 	}
 
 	@Nullable
 	@Override
 	public V getAndExpire(K key, long timeout, TimeUnit unit) {
-
-		return execute(new ValueDeserializingRedisCallback(key) {
-
-			@Override
-			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
-				return connection.getEx(rawKey, Expiration.from(timeout, unit));
-			}
-		});
+		return execute(
+				valueCallbackFor(key, (connection, rawKey) -> connection.getEx(rawKey, Expiration.from(timeout, unit))));
 	}
 
 	@Nullable
 	@Override
 	public V getAndExpire(K key, Duration timeout) {
-
-		return execute(new ValueDeserializingRedisCallback(key) {
-
-			@Override
-			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
-				return connection.getEx(rawKey, Expiration.from(timeout));
-			}
-		});
+		return execute(valueCallbackFor(key, (connection, rawKey) -> connection.getEx(rawKey, Expiration.from(timeout))));
 	}
 
 	@Nullable
 	@Override
 	public V getAndPersist(K key) {
-
-		return execute(new ValueDeserializingRedisCallback(key) {
-
-			@Override
-			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
-				return connection.getEx(rawKey, Expiration.persistent());
-			}
-		});
+		return execute(valueCallbackFor(key, (connection, rawKey) -> connection.getEx(rawKey, Expiration.persistent())));
 	}
 
 	@Override
 	public V getAndSet(K key, V newValue) {
 
 		byte[] rawValue = rawValue(newValue);
-		return execute(new ValueDeserializingRedisCallback(key) {
-
-			@Override
-			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
-				return connection.getSet(rawKey, rawValue);
-			}
-		});
+		return execute(valueCallbackFor(key, (connection, rawKey) -> connection.getSet(rawKey, rawValue)));
 	}
 
 	@Override
@@ -232,15 +193,10 @@ class DefaultValueOperations<K, V> extends AbstractOperations<K, V> implements V
 	@Override
 	public void set(K key, V value) {
 
+		byte[] rawKey = rawKey(key);
 		byte[] rawValue = rawValue(value);
-		execute(new ValueDeserializingRedisCallback(key) {
 
-			@Override
-			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
-				connection.set(rawKey, rawValue);
-				return null;
-			}
-		});
+		execute(connection -> connection.set(rawKey, rawValue));
 	}
 
 	@Override