diff --git a/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java b/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java index c4f08aab5..b3db62fd4 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java +++ b/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java @@ -147,36 +147,18 @@ public abstract class RedisConnectionUtils { return conn; } - private static void potentiallyRegisterTransactionSynchronisation(final RedisConnectionHolder connHolder, - RedisConnectionFactory factory) { + private static void potentiallyRegisterTransactionSynchronisation(RedisConnectionHolder connHolder, + final RedisConnectionFactory factory) { if (isActualNonReadonlyTransactionActive()) { - final RedisConnection connection = connHolder.getConnection(); - if (!connHolder.isTransactionSyncronisationActive()) { connHolder.setTransactionSyncronisationActive(true); - connection.multi(); + + RedisConnection conn = connHolder.getConnection(); + conn.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); - } - }); + TransactionSynchronizationManager.registerSynchronization(new RedisTransactionSynchronizer(connHolder, conn)); } } } @@ -202,9 +184,20 @@ public abstract class RedisConnectionUtils { * @param factory the Redis factory that the connection was created with */ public static void releaseConnection(RedisConnection conn, RedisConnectionFactory factory) { + if (conn == null) { return; } + + RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory); + + if (connHolder != null && connHolder.isTransactionSyncronisationActive()) { + if (log.isDebugEnabled()) { + log.debug("Redis Connection will be closed when transaction finished."); + } + return; + } + // Only release non-transactional/non-bound connections. if (!isConnectionTransactional(conn, factory)) { if (log.isDebugEnabled()) { @@ -220,12 +213,24 @@ public abstract class RedisConnectionUtils { * @param factory Redis factory */ public static void unbindConnection(RedisConnectionFactory factory) { + RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager .unbindResourceIfPossible(factory); + if (connHolder != null) { - RedisConnection connection = connHolder.getConnection(); - connection.close(); + if (connHolder.isTransactionSyncronisationActive()) { + if (log.isDebugEnabled()) { + log.debug("Redis Connection will be closed when outer transaction finished."); + } + } else { + if (log.isDebugEnabled()) { + log.debug("Closing bound connection."); + } + RedisConnection connection = connHolder.getConnection(); + connection.close(); + } } + } /** @@ -242,9 +247,60 @@ public abstract class RedisConnectionUtils { } RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager .getResource(connFactory); + return (connHolder != null && conn == connHolder.getConnection()); } + /** + * A {@link TransactionSynchronizationAdapter} that makes sure that the associated RedisConnection is released after the transaction completes. + * + * @author Christoph Strobl + * @author Thomas Darimont + */ + private static class RedisTransactionSynchronizer extends TransactionSynchronizationAdapter { + + private final RedisConnectionHolder connHolder; + private final RedisConnection connection; + + /** + * Creates a new {@link RedisTransactionSynchronizer}. + * + * @param connHolder + * @param connection + */ + private RedisTransactionSynchronizer(RedisConnectionHolder connHolder, RedisConnection connection) { + + this.connHolder = connHolder; + this.connection = connection; + } + + @Override + public void afterCompletion(int status) { + + try { + switch (status) { + + case TransactionSynchronization.STATUS_COMMITTED: + connection.exec(); + break; + + case TransactionSynchronization.STATUS_ROLLED_BACK: + case TransactionSynchronization.STATUS_UNKNOWN: + default: + connection.discard(); + } + } finally { + + if (log.isDebugEnabled()) { + log.debug("Closing bound connection after transaction completed with " + status); + } + + connHolder.setTransactionSyncronisationActive(false); + connection.close(); + } + } + } + /** * @author Christoph Strobl * @since 1.3 diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java index 6eb3c2ac3..fe7673a51 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java @@ -169,7 +169,13 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation RedisConnectionFactory factory = getConnectionFactory(); RedisConnection conn = null; try { - conn = RedisConnectionUtils.bindConnection(factory, enableTransactionSupport); + + if (enableTransactionSupport) { + // only bind resources in case of potential transaction synchronization + conn = RedisConnectionUtils.bindConnection(factory, enableTransactionSupport); + } else { + conn = RedisConnectionUtils.getConnection(factory); + } boolean existingConnection = TransactionSynchronizationManager.hasResource(factory); @@ -191,7 +197,12 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation // TODO: any other connection processing? return postProcessResult(result, connToUse, existingConnection); } finally { - RedisConnectionUtils.releaseConnection(conn, factory); + + if (enableTransactionSupport) { + RedisConnectionUtils.unbindConnection(factory); + } else { + RedisConnectionUtils.releaseConnection(conn, factory); + } } } diff --git a/src/test/java/org/springframework/data/redis/core/MultithreadedRedisTemplateTests.java b/src/test/java/org/springframework/data/redis/core/MultithreadedRedisTemplateTests.java new file mode 100644 index 000000000..6990cc501 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/core/MultithreadedRedisTemplateTests.java @@ -0,0 +1,92 @@ +/* + * 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 static org.junit.Assert.*; + +import java.util.Arrays; +import java.util.Collection; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.connection.srp.SrpConnectionFactory; + +/** + * @author Artem Bilian + * @author Christoph Strobl + */ +@RunWith(Parameterized.class) +public class MultithreadedRedisTemplateTests { + + private RedisConnectionFactory factory; + + public MultithreadedRedisTemplateTests(RedisConnectionFactory factory) { + this.factory = factory; + } + + @Parameters + public static Collection testParams() { + + JedisConnectionFactory jedis = new JedisConnectionFactory(); + jedis.setPort(6379); + jedis.afterPropertiesSet(); + + LettuceConnectionFactory lettuce = new LettuceConnectionFactory(); + lettuce.setPort(6379); + lettuce.afterPropertiesSet(); + + SrpConnectionFactory srp = new SrpConnectionFactory(); + srp.setPort(6479); + srp.afterPropertiesSet(); + + return Arrays.asList(new Object[][] { { jedis }, { lettuce }, { srp } }); + } + + /** + * @see DATAREDIS-300 + */ + @Test + public void assertResouresAreReleasedProperlyWhenSharingRedisTemplate() throws InterruptedException { + + final RedisTemplate template = new RedisTemplate(); + template.setConnectionFactory(factory); + template.afterPropertiesSet(); + + ExecutorService executor = Executors.newCachedThreadPool(); + + for (int i = 0; i < 9; i++) { + executor.execute(new Runnable() { + + @Override + public void run() { + template.boundValueOps("foo").get(); + } + }); + } + + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + +}