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

INT-4126: Fix Redis lock to use SET NX EX

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4126

Looks like `WATCH` doesn't provide good isolation when the call is performed concurrently at the same time from different Threads.

* Change locking algorithm as it is recommended by Redis `SET` command: http://redis.io/commands/set:
```
SET resource-name anystring NX EX max-lock-time
```

* Add `@Repeat(10)` to the `AggregatorWithRedisLocksTests#testDistributedAggregator()` since that helped to reproduce the issue

**Cherry-pick to 4.3.x & 4.2.x**
This commit is contained in:
Artem Bilan
2016-09-27 21:44:55 -04:00
parent 129984ec0a
commit 8aa9befef6
2 changed files with 25 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
View File

@@ -41,9 +41,8 @@ import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.data.redis.core.TimeoutUtils;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.data.redis.serializer.StringRedisSerializer;
@@ -420,32 +419,32 @@ public final class RedisLockRegistry implements LockRegistry {
Boolean success = false;
try {
success = RedisLockRegistry.this.redisTemplate.execute(new SessionCallback<Boolean>() {
@SuppressWarnings({ "unchecked", "rawtypes" })
success = RedisLockRegistry.this.redisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean execute(RedisOperations ops) throws DataAccessException {
String key = constructLockKey();
public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
ops.watch(key); //monitor key
/*
Perform Redis command 'SET resource-name anystring NX EX max-lock-time' directly.
As it is recommended by Redis: http://redis.io/commands/set.
This command isn't supported directly by RedisTemplate.
*/
long expireAfter = TimeoutUtils.toSeconds(RedisLockRegistry.this.expireAfter,
TimeUnit.MILLISECONDS);
RedisSerializer<String> serializer = RedisLockRegistry.this.redisTemplate.getStringSerializer();
byte[][] actualArgs = new byte[][] {
serializer.serialize(constructLockKey()),
RedisLockRegistry.this.lockSerializer.serialize(RedisLock.this),
serializer.serialize("NX"),
serializer.serialize("EX"),
serializer.serialize(String.valueOf(expireAfter))
};
if (ops.opsForValue().get(key) != null) {
ops.unwatch(); //key already exists, stop monitoring
return false;
}
ops.multi(); //transaction start
//set the value and expire
ops.opsForValue()
.set(key, RedisLock.this, RedisLockRegistry.this.expireAfter, TimeUnit.MILLISECONDS);
//exec will contain all operations result or null - if execution has been aborted due to 'watch'
return ops.exec() != null;
return connection.execute("SET", actualArgs) != null;
}
});
}
finally {

5
spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java
View File

@@ -46,11 +46,13 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.annotation.Repeat;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @author Artem Bilan
* @since 4.0
*
*/
@@ -125,6 +127,7 @@ public class AggregatorWithRedisLocksTests extends RedisAvailableTests {
@Test
@RedisAvailable
@Repeat(10)
public void testDistributedAggregator() throws Exception {
this.releaseStrategy.reset(1);
Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1));
@@ -223,5 +226,7 @@ public class AggregatorWithRedisLocksTests extends RedisAvailableTests {
this.callers = new AtomicInteger();
this.maxCallers = new AtomicInteger();
}
}
}