Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-73 - Add support for spring managed transactions.

Browse Source 
RedisTemplate now supports enabling of transaction support, which is disabled by default.

In case of enabled transaction support the RedisConnections will be bound to the current thread during ongoing outer transactions. We call MULTI at the beginning and depending on transaction state EXEC or DISCARD at its end.

To support read operations during the transaction we wrap a proxy around the bound connection piping read operations to a new (non thread bound) connection obtained by the underlying RedisConnectionFactory.

Transaction support is available for jedis, lettuce and srp, while had to be skipped for jredis due to the lack of support for MULTI.

Original pull request: #64.
This commit is contained in:
Christoph Strobl
2014-03-20 14:55:24 +01:00
committed by Thomas Darimont
parent 4e4a2961da
commit bc1bbd1d39
11 changed files with 1080 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

365
src/main/java/org/springframework/data/redis/core/RedisCommand.java Normal file
View File

@@ -0,0 +1,365 @@
/*
* Copyright 2014 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
*
* http://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.data.redis.core;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.jredis.Redis;
import org.springframework.util.StringUtils;
/**
* @author Christoph Strobl
* @author Thomas Darimont
* @since 1.3
* @see Redis command list:
* https://github.com/antirez/redis/blob/93e7a130fc9594e41ccfc996b5eca7626ae5356a/src/redis.c#L119
*/
enum RedisCommand {
// -- A
APPEND("rw", 2, 2), //
AUTH("rw", 1, 1), //
// -- B
BGREWRITEAOF("r", 0, 0, "bgwriteaof"), //
BGSAVE("r", 0, 0), //
BITCOUNT("r", 1, 3), //
BITOP("rw", 3), //
BITPOS("r", 2, 4), //
BLPOP("rw", 2), //
BRPOP("rw", 2), //
BRPOPLPUSH("rw", 3), //
// -- C
CLIENT_KILL("rw", 1, 1), //
CLIENT_LIST("r", 0, 0), //
CLIENT_GETNAME("r", 0, 0), //
CLIENT_PAUSE("rw", 1, 1), //
CLIENT_SETNAME("w", 1, 1), //
CONFIG_GET("r", 1, 1, "getconfig"), //
CONFIG_REWRITE("rw", 0, 0), //
CONFIG_SET("w", 2, 2, "setconfig"), //
CONFIG_RESETSTAT("w", 0, 0, "resetconfigstats"), //
// -- D
DBSIZE("r", 0, 0), //
DECR("w", 1, 1), //
DECRBY("w", 2, 2), //
DEL("rw", 1), //
DISCARD("rw", 0, 0), //
DUMP("r", 1, 1), //
// -- E
ECHO("r", 1, 1), //
EVAL("rw", 2), //
EVALSHA("rw", 2), //
EXEC("rw", 0, 0), //
EXISTS("r", 1, 1), //
EXPIRE("rw", 2, 2), //
EXPIREAT("rw", 2, 2), //
// -- F
FLUSHALL("w", 0, 0), //
FLUSHDB("w", 0, 0), //
// -- G
GET("r", 1, 1), //
GETBIT("r", 2, 2), //
GETRANGE("r", 3, 3), //
GETSET("rw", 2, 2), //
// -- H
HDEL("rw", 2), //
HEXISTS("r", 2, 2), //
HGET("r", 2, 2), //
HGETALL("r", 1, 1), //
HINCRBY("rw", 3, 3), //
HINCBYFLOAT("rw", 3, 3), //
HKEYS("r", 1), //
HLEN("r", 1), //
HMGET("r", 2), //
HMSET("w", 3), //
HSET("w", 3, 3), //
HSETNX("w", 3, 3), //
HVALS("r", 1, 1), //
// -- I
INCR("rw", 1), //
INCRBYFLOAT("rw", 2, 2), //
INFO("r", 0), //
// -- K
KEYS("r", 1), //
// -- L
LASTSAVE("r", 0), //
LINDEX("r", 2, 2), //
LINSERT("rw", 4, 4), //
LLEN("r", 1, 1), //
LPOP("rw", 1, 1), //
LPUSH("rw", 2), //
LPUSHX("rw", 2), //
LRANGE("r", 3, 3), //
LREM("rw", 3, 3), //
LSET("w", 3, 3), //
LTRIM("w", 3, 3), //
// -- M
MGET("r", 1), //
MIGRATE("rw", 0), //
MONITOR("rw", 0, 0), //
MOVE("rw", 2, 2), //
MSET("w", 2), //
MSETNX("w", 2), //
MULTI("rw", 0, 0), //
// -- P
PERSIST("rw", 1, 1), //
PEXPIRE("rw", 2, 2), //
PEXPIREAT("rw", 2, 2), //
PING("r", 0, 0), //
PSETEX("w", 3), //
PSUBSCRIBE("r", 1), //
PTTL("r", 1, 1), //
// -- Q
QUIT("rw", 0, 0), //
// -- R
RANDOMKEY("r", 0, 0), //
RANAME("w", 2, 2), //
RENAMENX("w", 2, 2), //
RESTORE("w", 3, 3), //
RPOP("rw", 1, 1), //
RPOPLPUSH("rw", 2, 2), //
RPUSH("rw", 2), //
RPUSHX("rw", 2, 2), //
// -- S
SADD("rw", 2), //
SAVE("rw", 0, 0), //
SCARD("r", 1, 1), //
SCRIPT_EXISTS("r", 1), //
SCRIPT_FLUSH("rw", 0, 0), //
SCRIPT_KILL("rw", 0, 0), //
SCRIPT_LOAD("rw", 1, 1), //
SDIFF("r", 1), //
SDIFFSTORE("rw", 2), //
SELECT("rw", 0, 0), //
SET("w", 2), //
SETBIT("rw", 3, 3), //
SETEX("w", 3, 3), //
SETNX("w", 2, 2), //
SETRANGE("rw", 3, 3), //
SHUTDOWN("rw", 0), //
SINTER("r", 1), //
SINTERSTORE("rw", 2), //
SISMEMBER("r", 2), //
SLAVEOF("w", 2), //
SLOWLOG("rw", 1), //
SMEMBERS("r", 1, 1), //
SMOVE("rw", 3, 3), //
SORT("rw", 1), //
SPOP("rw", 1, 1), //
SRANDMEMBER("r", 1, 1), //
SREM("rw", 2), //
STRLEN("r", 1, 1), //
SUBSCRIBE("rw", 1), //
SUNION("r", 1), //
SUNIONSTORE("rw ", 2), //
SYNC("rw", 0, 0), //
// -- T
TIME("r", 0, 0), //
TTL("r", 1, 1), //
TYPE("r", 1, 1), //
// -- U
UNSUBSCRIBE("rw", 0), //
UNWATCH("rw", 0, 0), //
// -- W
WATCH("rw", 1), //
// -- Z
ZADD("rw", 3), //
ZCARD("r", 1), //
ZCOUNT("r", 3, 3), //
ZINCRBY("rw", 3), //
ZINTERSTORE("rw", 3), //
ZRANGE("r", 3), //
ZRANGEBYSCORE("r", 3), //
ZRANK("r", 2, 2), //
ZREM("rw", 2), //
ZREMRANGEBYRANK("rw", 3, 3), //
ZREMRANGEBYSCORE("rm", 3, 3), //
ZREVRANGE("r", 3), //
ZREVRANGEBYSCORE("r", 3), //
ZREVRANK("r", 2, 2), //
ZSCORE("r", 2, 2), //
ZUNIONSTORE("rw", 3), //
SCAN("r", 1), //
SSCAN("r", 2), //
HSCAN("r", 2), //
ZSCAN("r", 2), //
// -- UNKNOWN / DEFAULT
UNKNOWN("rw", -1);
private boolean read = true;
private boolean write = true;
private Set<String> alias = new HashSet<String>(1);
private int minArgs = -1;
private int maxArgs = -1;
private final static Map<String, RedisCommand> commandLookup;
static {
commandLookup = buildCommandLookupTable();
}
private static Map<String, RedisCommand> buildCommandLookupTable() {
RedisCommand[] cmds = RedisCommand.values();
Map<String, RedisCommand> map = new HashMap<String, RedisCommand>(cmds.length, 1.0F);
for (RedisCommand cmd : cmds) {
map.put(cmd.name().toLowerCase(), cmd);
for (String alias : cmd.alias) {
map.put(alias, cmd);
}
}
return Collections.unmodifiableMap(map);
}
private RedisCommand(String mode, int minArgs) {
this(mode, minArgs, -1);
}
private RedisCommand(String mode, int minArgs, int maxArgs) {
if (StringUtils.hasText(mode)) {
this.read = mode.toLowerCase().indexOf('r') > -1;
this.write = mode.toLowerCase().indexOf('w') > -1;
}
this.minArgs = minArgs;
this.maxArgs = maxArgs;
}
/**
* Creates a new {@link RedisCommand}.
*
* @param mode
* @param minArgs
* @param maxArgs
* @param alias
*/
private RedisCommand(String mode, int minArgs, int maxArgs, String... alias) {
this(mode, minArgs, maxArgs);
if (alias != null && alias.length > 0) {
this.alias.addAll(Arrays.asList(alias));
}
}
/**
* @return {@literal true} if the command requires arguments
*/
public boolean requiresArguments() {
return minArgs >= 0;
}
/**
* @return {@literal true} if an exact number of arguments is expected
*/
public boolean requiresExactNumberOfArguments() {
return maxArgs >= 0;
}
/**
* @return {@literal true} if the command triggers a read operation
*/
public boolean isRead() {
return read;
}
/**
* @return {@literal true} if the command triggers a write operation
*/
public boolean isWrite() {
return write;
}
/**
* @return {@literal true} if values are read but not written
*/
public boolean isReadonly() {
return read && !write;
}
/**
* {@link String#equalsIgnoreCase(String)} compare the given string representation of {@literal command} against the
* {@link #toString()} representation of the command as well as its given {@link #alias}.
*
* @param command
* @return true if positive match.
*/
public boolean isRepresentedBy(String command) {
if (!StringUtils.hasText(command)) {
return false;
}
if (toString().equalsIgnoreCase(command)) {
return true;
}
return alias.contains(command.toLowerCase());
}
/**
* Validates given argument count against expected ones.
*
* @param nrArguments
* @throws {@link IllegalArgumentException} in case argument count does not match expected.
*/
public void validateArgumentCount(int nrArguments) {
if (requiresArguments()) {
if (requiresExactNumberOfArguments()) {
if (nrArguments != maxArgs) {
throw new IllegalArgumentException(String.format("%s command requires %s arguments.", this.name(),
this.maxArgs));
}
}
if (nrArguments < minArgs) {
throw new IllegalArgumentException(String.format("%s command requires at least %s arguments.", this.name(),
this.minArgs));
}
}
}
/**
* Returns the command represented by the given {@code key}. Returns {@link #UNKNOWN} if no matching command could be
* found.
*
* @param key
* @return
*/
public static RedisCommand failsafeCommandLookup(String key) {
if (!StringUtils.hasText(key)) {
return RedisCommand.UNKNOWN;
}
RedisCommand cmd = commandLookup.get(key.toLowerCase());
if (cmd != null) {
return cmd;
}
return RedisCommand.UNKNOWN;
}
}

210
src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2011-2013 the original author or authors.
* Copyright 2011-2014 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.
@@ -15,11 +15,20 @@
*/
package org.springframework.data.redis.core;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.cglib.proxy.MethodProxy;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.transaction.support.ResourceHolder;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;
@@ -28,6 +37,8 @@ import org.springframework.util.Assert;
* 'transactions'/scopes.
*
* @author Costin Leau
* @author Christoph Strobl
* @author Thomas Darimont
*/
public abstract class RedisConnectionUtils {
@@ -37,10 +48,22 @@ public abstract class RedisConnectionUtils {
* Binds a new Redis connection (from the given factory) to the current thread, if none is already bound.
*
* @param factory connection factory
* @return a new Redis connection
* @return a new Redis connection without transaction support.
*/
public static RedisConnection bindConnection(RedisConnectionFactory factory) {
return doGetConnection(factory, true, true);
return bindConnection(factory, false);
}
/**
* Binds a new Redis connection (from the given factory) to the current thread, if none is already bound and enables
* transaction support if {@code enableTranactionSupport} is set to {@literal true}.
*
* @param factory connection factory
* @param enableTranactionSupport
* @return a new Redis connection with transaction support if requested.
*/
public static RedisConnection bindConnection(RedisConnectionFactory factory, boolean enableTranactionSupport) {
return doGetConnection(factory, true, true, enableTranactionSupport);
}
/**
@@ -49,10 +72,23 @@ public abstract class RedisConnectionUtils {
* otherwise.
*
* @param factory connection factory for creating the connection
* @return an active Redis connection
* @return an active Redis connection without transaction management.
*/
public static RedisConnection getConnection(RedisConnectionFactory factory) {
return doGetConnection(factory, true, false);
return getConnection(factory, false);
}
/**
* Gets a Redis connection from the given factory. Is aware of and will return any existing corresponding connections
* bound to the current thread, for example when using a transaction manager. Will always create a new connection
* otherwise.
*
* @param factory connection factory for creating the connection
* @param enableTranactionSupport
* @return an active Redis connection with transaction management if requested.
*/
public static RedisConnection getConnection(RedisConnectionFactory factory, boolean enableTranactionSupport) {
return doGetConnection(factory, true, false, enableTranactionSupport);
}
/**
@@ -64,34 +100,100 @@ public abstract class RedisConnectionUtils {
* @param allowCreate whether a new (unbound) connection should be created when no connection can be found for the
* current thread
* @param bind binds the connection to the thread, in case one was created
* @param enableTransactionSupport
* @return an active Redis connection
*/
public static RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind) {
public static RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind,
boolean enableTransactionSupport) {
Assert.notNull(factory, "No RedisConnectionFactory specified");
RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory);
// TODO: investigate tx synchronization
if (connHolder != null)
if (connHolder != null) {
if (enableTransactionSupport) {
potentiallyRegisterTransactionSynchronisation(connHolder, factory);
}
return connHolder.getConnection();
}
if (!allowCreate) {
throw new IllegalArgumentException("No connection found and allowCreate = false");
}
if (log.isDebugEnabled())
if (log.isDebugEnabled()) {
log.debug("Opening RedisConnection");
}
RedisConnection conn = factory.getConnection();
if (bind) {
connHolder = new RedisConnectionHolder(conn);
RedisConnection connectionToBind = conn;
if (enableTransactionSupport && isActualNonReadonlyTransactionActive()) {
connectionToBind = createConnectionProxy(conn, factory);
}
connHolder = new RedisConnectionHolder(connectionToBind);
TransactionSynchronizationManager.bindResource(factory, connHolder);
if (enableTransactionSupport) {
potentiallyRegisterTransactionSynchronisation(connHolder, factory);
}
return connHolder.getConnection();
}
return conn;
}
private static void potentiallyRegisterTransactionSynchronisation(final RedisConnectionHolder connHolder,
RedisConnectionFactory factory) {
if (isActualNonReadonlyTransactionActive()) {
final RedisConnection connection = connHolder.getConnection();
if (!connHolder.isTransactionSyncronisationActive()) {
connHolder.setTransactionSyncronisationActive(true);
connection.multi();
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
@Override
public void afterCompletion(int status) {
switch (status) {
case TransactionSynchronization.STATUS_COMMITTED:
connection.exec();
break;
case TransactionSynchronization.STATUS_ROLLED_BACK:
case TransactionSynchronization.STATUS_UNKNOWN:
default:
connection.discard();
}
connHolder.setTransactionSyncronisationActive(false);
}
});
}
}
}
private static boolean isActualNonReadonlyTransactionActive() {
return TransactionSynchronizationManager.isActualTransactionActive()
&& !TransactionSynchronizationManager.isCurrentTransactionReadOnly();
}
private static RedisConnection createConnectionProxy(RedisConnection connection, RedisConnectionFactory factory) {
ProxyFactory proxyFactory = new ProxyFactory(connection);
proxyFactory.addAdvice(new ConnectionSplittingInterceptor(factory));
return RedisConnection.class.cast(proxyFactory.getProxy());
}
/**
* Closes the given connection, created via the given factory if not managed externally (i.e. not bound to the
* thread).
@@ -143,17 +245,83 @@ public abstract class RedisConnectionUtils {
return (connHolder != null && conn == connHolder.getConnection());
}
/**
* @author Christoph Strobl
* @since 1.3
*/
static class ConnectionSplittingInterceptor implements MethodInterceptor,
org.springframework.cglib.proxy.MethodInterceptor {
private final RedisConnectionFactory factory;
public ConnectionSplittingInterceptor(RedisConnectionFactory factory) {
this.factory = factory;
}
@Override
public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable {
RedisCommand commandToExecute = RedisCommand.failsafeCommandLookup(method.getName());
if (isPotentiallyThreadBoundCommand(commandToExecute)) {
if (log.isDebugEnabled()) {
log.debug(String.format("Invoke '%s' on bound conneciton", method.getName()));
}
return invoke(method, obj, args);
}
if (log.isDebugEnabled()) {
log.debug(String.format("Invoke '%s' on unbound conneciton", method.getName()));
}
RedisConnection connection = factory.getConnection();
try {
return invoke(method, connection, args);
} finally {
// properly close the unbound connection after executing command
if (!connection.isClosed()) {
connection.close();
}
}
}
private Object invoke(Method method, Object target, Object[] args) throws Throwable {
try {
return method.invoke(target, args);
} catch (InvocationTargetException e) {
throw e.getCause();
}
}
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
return intercept(invocation.getThis(), invocation.getMethod(), invocation.getArguments(), null);
}
private boolean isPotentiallyThreadBoundCommand(RedisCommand command) {
return RedisCommand.UNKNOWN.equals(command) || !command.isReadonly();
}
}
/**
* @author Christoph Strobl
*/
private static class RedisConnectionHolder implements ResourceHolder {
private boolean isVoid = false;
private boolean unbound;
private final RedisConnection conn;
private boolean transactionSyncronisationActive;
public RedisConnectionHolder(RedisConnection conn) {
this.conn = conn;
}
public boolean isVoid() {
return isVoid;
return unbound;
}
public RedisConnection getConnection() {
@@ -165,7 +333,23 @@ public abstract class RedisConnectionUtils {
}
public void unbound() {
this.isVoid = true;
this.unbound = true;
}
/**
* @return
* @since 1.3
*/
public boolean isTransactionSyncronisationActive() {
return transactionSyncronisationActive;
}
/**
* @param transactionSyncronisationActive
* @since 1.3
*/
public void setTransactionSyncronisationActive(boolean transactionSyncronisationActive) {
this.transactionSyncronisationActive = transactionSyncronisationActive;
}
}
}

17
src/main/java/org/springframework/data/redis/core/RedisTemplate.java
View File

@@ -77,6 +77,7 @@ import org.springframework.util.CollectionUtils;
*/
public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperations<K, V> {
private boolean enableTransactionSupport = false;
private boolean exposeConnection = false;
private boolean initialized = false;
private boolean enableDefaultSerializer = true;
@@ -168,7 +169,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
RedisConnectionFactory factory = getConnectionFactory();
RedisConnection conn = null;
try {
conn = RedisConnectionUtils.getConnection(factory);
conn = RedisConnectionUtils.bindConnection(factory, enableTransactionSupport);
boolean existingConnection = TransactionSynchronizationManager.hasResource(factory);
@@ -200,7 +201,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
RedisConnectionFactory factory = getConnectionFactory();
// bind connection
RedisConnectionUtils.bindConnection(factory);
RedisConnectionUtils.bindConnection(factory, enableTransactionSupport);
try {
return session.execute(this);
} finally {
@@ -218,7 +219,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
RedisConnectionFactory factory = getConnectionFactory();
// bind connection
RedisConnectionUtils.bindConnection(factory);
RedisConnectionUtils.bindConnection(factory, enableTransactionSupport);
try {
return execute(new RedisCallback<List<Object>>() {
public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
@@ -1053,4 +1054,14 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
}
});
}
/**
* If set to {@code true} {@link RedisTemplate} will use {@literal MULTI...EXEC|DISCARD} to keep track of operations.
*
* @param enableTransactionSupport
* @since 1.3
*/
public void setEnableTransactionSupport(boolean enableTransactionSupport) {
this.enableTransactionSupport = enableTransactionSupport;
}
}