INT-4248: Refactor RedisLockRegistry
JIRA: https://jira.spring.io/browse/INT-4248 To avoid unexpected double locking behavior in the cluster, remove the local cache functionality. Now with the new `clientId` property, the `expire` for the record in store is always update on each lock operation
This commit is contained in:
Vedran Pavic
committed by
Artem Bilan
Artem Bilan
parent
28058884f4
commit
f7f7bdd067
@@ -16,36 +16,28 @@
|
||||
|
||||
package org.springframework.integration.redis.util;
|
||||
|
||||
import java.net.InetAddress;
|
||||
import java.net.UnknownHostException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.time.Instant;
|
||||
import java.time.ZoneId;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.text.SimpleDateFormat;
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.WeakHashMap;
|
||||
import java.util.Date;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.locks.Condition;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.dao.CannotAcquireLockException;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.core.RedisCallback;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
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;
|
||||
import org.springframework.integration.support.locks.DefaultLockRegistry;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.core.script.DefaultRedisScript;
|
||||
import org.springframework.data.redis.core.script.RedisScript;
|
||||
import org.springframework.integration.support.locks.ExpirableLockRegistry;
|
||||
import org.springframework.integration.support.locks.LockRegistry;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
@@ -66,61 +58,48 @@ import org.springframework.util.Assert;
|
||||
* <p>
|
||||
* <b>Note: This is not intended for low latency applications.</b> It is intended
|
||||
* for resource locking across multiple JVMs.
|
||||
* When a lock is released by a remote system, waiting threads may take up to 100ms
|
||||
* to acquire the lock.
|
||||
* A more performant version would need to get notifications from the Redis stores
|
||||
* of key changes. This is currently only available using the SYNC command.
|
||||
* <p>
|
||||
* This limitation will usually not apply when a lock is released within this registry,
|
||||
* unless another system takes the lock after the local lock is acquired here.
|
||||
* A {@link DefaultLockRegistry} is used internally to achieve this optimization.
|
||||
* <p>
|
||||
* {@link Condition}s are not supported.
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @author Konstantin Yakimov
|
||||
* @author Artem Bilan
|
||||
* @author Vedran Pavic
|
||||
*
|
||||
* @since 4.0
|
||||
*
|
||||
*/
|
||||
public final class RedisLockRegistry implements LockRegistry {
|
||||
public final class RedisLockRegistry implements ExpirableLockRegistry {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(RedisLockRegistry.class);
|
||||
|
||||
private static final byte[] hostName;
|
||||
|
||||
private static final long DEFAULT_EXPIRE_AFTER = 60000;
|
||||
|
||||
private static final String OBTAIN_LOCK_SCRIPT =
|
||||
"local lockClientId = redis.call('GET', KEYS[1])\n" +
|
||||
"if lockClientId == ARGV[1] then\n" +
|
||||
" redis.call('PEXPIRE', KEYS[1], ARGV[2])\n" +
|
||||
" return true\n" +
|
||||
"elseif not lockClientId then\n" +
|
||||
" redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])\n" +
|
||||
" return true\n" +
|
||||
"end\n" +
|
||||
"return false";
|
||||
|
||||
private final ConcurrentMap<String, RedisLock> locks = new ConcurrentHashMap<>();
|
||||
|
||||
private final String clientId = UUID.randomUUID().toString();
|
||||
|
||||
private final String registryKey;
|
||||
|
||||
private final RedisTemplate<String, RedisLock> redisTemplate;
|
||||
private final StringRedisTemplate redisTemplate;
|
||||
|
||||
private final ThreadLocal<Set<RedisLock>> weakThreadLocks = new ThreadLocal<Set<RedisLock>>();
|
||||
|
||||
private final ThreadLocal<List<RedisLock>> hardThreadLocks = new ThreadLocal<List<RedisLock>>();
|
||||
private final RedisScript<Boolean> obtainLockScript;
|
||||
|
||||
private final long expireAfter;
|
||||
|
||||
private final LockRegistry localRegistry;
|
||||
|
||||
private final LockSerializer lockSerializer = new LockSerializer();
|
||||
|
||||
private boolean useWeakReferences = false;
|
||||
|
||||
static {
|
||||
String host;
|
||||
try {
|
||||
host = InetAddress.getLocalHost().getHostName();
|
||||
}
|
||||
catch (UnknownHostException e) {
|
||||
host = "unknownHost";
|
||||
}
|
||||
hostName = host.getBytes();
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructs a lock registry with the default (60 second) lock expiration and a default
|
||||
* local {@link DefaultLockRegistry}.
|
||||
* Constructs a lock registry with the default (60 second) lock expiration.
|
||||
* @param connectionFactory The connection factory.
|
||||
* @param registryKey The key prefix for locks.
|
||||
*/
|
||||
@@ -129,211 +108,68 @@ public final class RedisLockRegistry implements LockRegistry {
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructs a lock registry with the supplied lock expiration and a default
|
||||
* local {@link DefaultLockRegistry}.
|
||||
* Constructs a lock registry with the supplied lock expiration.
|
||||
* @param connectionFactory The connection factory.
|
||||
* @param registryKey The key prefix for locks.
|
||||
* @param expireAfter The expiration in milliseconds.
|
||||
*/
|
||||
public RedisLockRegistry(RedisConnectionFactory connectionFactory, String registryKey, long expireAfter) {
|
||||
this(connectionFactory, registryKey, expireAfter, new DefaultLockRegistry());
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructs a lock registry with the supplied lock expiration and a custom local {@link LockRegistry}.
|
||||
* @param connectionFactory The connection factory.
|
||||
* @param registryKey The key prefix for locks.
|
||||
* @param expireAfter The expiration in milliseconds.
|
||||
* @param localRegistry The local registry used to reduce wait time,
|
||||
* {@link DefaultLockRegistry} is used by default.
|
||||
*/
|
||||
public RedisLockRegistry(RedisConnectionFactory connectionFactory, String registryKey,
|
||||
long expireAfter, LockRegistry localRegistry) {
|
||||
Assert.notNull(connectionFactory, "'connectionFactory' cannot be null");
|
||||
Assert.notNull(registryKey, "'registryKey' cannot be null");
|
||||
Assert.notNull(localRegistry, "'localRegistry' cannot be null");
|
||||
this.redisTemplate = new RedisTemplate<>();
|
||||
this.redisTemplate.setConnectionFactory(connectionFactory);
|
||||
this.redisTemplate.setKeySerializer(new StringRedisSerializer());
|
||||
this.redisTemplate.setValueSerializer(this.lockSerializer);
|
||||
this.redisTemplate.afterPropertiesSet();
|
||||
this.redisTemplate = new StringRedisTemplate(connectionFactory);
|
||||
this.obtainLockScript = new DefaultRedisScript<>(OBTAIN_LOCK_SCRIPT, Boolean.class);
|
||||
this.registryKey = registryKey;
|
||||
this.expireAfter = expireAfter;
|
||||
this.localRegistry = localRegistry;
|
||||
}
|
||||
|
||||
/**
|
||||
* Change the state of thread local weak references storage for unlocked locks.
|
||||
* Thread local weak references are used for lock obtaining optimization -
|
||||
* thread will get same {@link RedisLock} object for certain key before actual
|
||||
* locking and after unlocking (if variable still exists).
|
||||
* <p>While is switched off (by default) every {@link RedisLockRegistry#obtain} call will provide
|
||||
* different {@link RedisLock} objects for same unlocked key.
|
||||
* @param useWeakReferences set to true for switch thread local weak references storage on, false by default
|
||||
* @since 4.0.7
|
||||
*/
|
||||
public void setUseWeakReferences(boolean useWeakReferences) {
|
||||
this.useWeakReferences = useWeakReferences;
|
||||
}
|
||||
|
||||
/**
|
||||
* Weak referenced locks, lock is kept here when actual lock is NOT gained.
|
||||
* Used for obtaining same lock object within same thread and key.
|
||||
* To avoid memory leaks lock objects without actual lock are kept as weak references.
|
||||
* After gaining the actual lock, lock object moves from weak reference to hard reference and vise a versa.
|
||||
*/
|
||||
private Collection<RedisLock> getWeakThreadLocks() {
|
||||
Set<RedisLock> locks = this.weakThreadLocks.get();
|
||||
if (locks == null) {
|
||||
locks = Collections.newSetFromMap(new WeakHashMap<RedisLock, Boolean>());
|
||||
this.weakThreadLocks.set(locks);
|
||||
}
|
||||
return locks;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hard referenced locks, lock is kept here when actual lock is gained.
|
||||
*/
|
||||
private Collection<RedisLock> getHardThreadLocks() {
|
||||
List<RedisLock> locks = this.hardThreadLocks.get();
|
||||
if (locks == null) {
|
||||
locks = new LinkedList<>();
|
||||
this.hardThreadLocks.set(locks);
|
||||
}
|
||||
return locks;
|
||||
}
|
||||
|
||||
private RedisLock findLock(Collection<RedisLock> locks, Object key) {
|
||||
if (locks != null) {
|
||||
for (RedisLock lock : locks) {
|
||||
if (lock.getLockKey().equals(key)) {
|
||||
return lock;
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private void toHardThreadStorage(RedisLock lock) {
|
||||
if (this.weakThreadLocks.get() != null) {
|
||||
this.weakThreadLocks.get().remove(lock);
|
||||
}
|
||||
|
||||
getHardThreadLocks().add(lock);
|
||||
|
||||
//clean up
|
||||
if (this.weakThreadLocks.get() != null && this.weakThreadLocks.get().isEmpty()) {
|
||||
this.weakThreadLocks.remove();
|
||||
}
|
||||
}
|
||||
|
||||
private void toWeakThreadStorage(RedisLock lock) {
|
||||
//to avoid collection creation on existence check use direct fields
|
||||
if (this.hardThreadLocks.get() != null) {
|
||||
getHardThreadLocks().remove(lock);
|
||||
}
|
||||
|
||||
if (this.useWeakReferences) {
|
||||
getWeakThreadLocks().add(lock);
|
||||
}
|
||||
|
||||
//clean up
|
||||
if (this.hardThreadLocks.get() != null && this.hardThreadLocks.get().isEmpty()) {
|
||||
this.hardThreadLocks.remove();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Lock obtain(Object lockKey) {
|
||||
Assert.isInstanceOf(String.class, lockKey);
|
||||
|
||||
//try to find the lock within hard references
|
||||
RedisLock lock = findLock(this.hardThreadLocks.get(), lockKey);
|
||||
|
||||
/*
|
||||
* If the lock is locked, check that it matches what's in the store.
|
||||
* If it doesn't, the lock must have expired.
|
||||
*/
|
||||
if (lock != null && lock.thread != null) {
|
||||
RedisLock lockInStore = this.redisTemplate.boundValueOps(this.registryKey + ":" + lockKey).get();
|
||||
if (lockInStore == null || !lock.equals(lockInStore)) {
|
||||
try {
|
||||
lock.unlock();
|
||||
}
|
||||
catch (Exception e) {
|
||||
if (logger.isWarnEnabled()) {
|
||||
logger.warn("Lock was released due to expiration. A new one will be obtained...", e);
|
||||
}
|
||||
}
|
||||
if (this.hardThreadLocks.get() != null) {
|
||||
this.hardThreadLocks.get().remove(lock);
|
||||
}
|
||||
if (this.weakThreadLocks.get() != null) {
|
||||
this.weakThreadLocks.get().remove(lock);
|
||||
}
|
||||
lock = null;
|
||||
}
|
||||
}
|
||||
|
||||
if (lock == null) {
|
||||
//try to find the lock within weak references
|
||||
lock = findLock(this.weakThreadLocks.get(), lockKey);
|
||||
|
||||
if (lock == null) {
|
||||
lock = new RedisLock((String) lockKey);
|
||||
|
||||
if (this.useWeakReferences) {
|
||||
getWeakThreadLocks().add(lock);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return lock;
|
||||
String path = (String) lockKey;
|
||||
return this.locks.computeIfAbsent(path, RedisLock::new);
|
||||
}
|
||||
|
||||
public Collection<Lock> listLocks() {
|
||||
return this.redisTemplate.execute((RedisCallback<Collection<Lock>>) connection -> {
|
||||
Set<byte[]> keys = connection.keys((RedisLockRegistry.this.registryKey + ":*").getBytes());
|
||||
if (keys.size() > 0) {
|
||||
List<byte[]> locks = connection.mGet(keys.toArray(new byte[keys.size()][]));
|
||||
return locks.stream()
|
||||
.map(RedisLockRegistry.this.lockSerializer::deserialize)
|
||||
.collect(Collectors.toList());
|
||||
@Override
|
||||
public void expireUnusedOlderThan(long age) {
|
||||
synchronized (this.locks) {
|
||||
Iterator<Map.Entry<String, RedisLock>> iterator = this.locks.entrySet().iterator();
|
||||
long now = System.currentTimeMillis();
|
||||
while (iterator.hasNext()) {
|
||||
Map.Entry<String, RedisLock> entry = iterator.next();
|
||||
RedisLock lock = entry.getValue();
|
||||
if (now - lock.getLockedAt() > age && !lock.isAcquiredInThisProcess()) {
|
||||
iterator.remove();
|
||||
}
|
||||
}
|
||||
return Collections.emptyList();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private final class RedisLock implements Lock {
|
||||
|
||||
private final String lockKey;
|
||||
|
||||
private long lockedAt;
|
||||
private final ReentrantLock localLock = new ReentrantLock();
|
||||
|
||||
private Thread thread;
|
||||
private volatile long lockedAt;
|
||||
|
||||
private String threadName;
|
||||
|
||||
private byte[] lockHost;
|
||||
|
||||
private int reLock;
|
||||
|
||||
RedisLock(String lockKey) {
|
||||
this.lockKey = lockKey;
|
||||
this.lockHost = RedisLockRegistry.hostName;
|
||||
private RedisLock(String path) {
|
||||
this.lockKey = constructLockKey(path);
|
||||
}
|
||||
|
||||
private String getLockKey() {
|
||||
return this.lockKey;
|
||||
private String constructLockKey(String path) {
|
||||
return RedisLockRegistry.this.registryKey + ":" + path;
|
||||
}
|
||||
|
||||
public long getLockedAt() {
|
||||
return this.lockedAt;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void lock() {
|
||||
Lock localLock = RedisLockRegistry.this.localRegistry.obtain(this.lockKey);
|
||||
localLock.lock();
|
||||
this.localLock.lock();
|
||||
while (true) {
|
||||
try {
|
||||
while (!this.obtainLock()) {
|
||||
while (!obtainLock()) {
|
||||
Thread.sleep(100); //NOSONAR
|
||||
}
|
||||
break;
|
||||
@@ -346,7 +182,7 @@ public final class RedisLockRegistry implements LockRegistry {
|
||||
*/
|
||||
}
|
||||
catch (Exception e) {
|
||||
localLock.unlock();
|
||||
this.localLock.unlock();
|
||||
rethrowAsLockException(e);
|
||||
}
|
||||
}
|
||||
@@ -358,183 +194,104 @@ public final class RedisLockRegistry implements LockRegistry {
|
||||
|
||||
@Override
|
||||
public void lockInterruptibly() throws InterruptedException {
|
||||
Lock localLock = RedisLockRegistry.this.localRegistry.obtain(this.lockKey);
|
||||
localLock.lockInterruptibly();
|
||||
this.localLock.lockInterruptibly();
|
||||
try {
|
||||
while (!this.obtainLock()) {
|
||||
while (!obtainLock()) {
|
||||
Thread.sleep(100); //NOSONAR
|
||||
}
|
||||
}
|
||||
catch (InterruptedException ie) {
|
||||
localLock.unlock();
|
||||
this.localLock.unlock();
|
||||
Thread.currentThread().interrupt();
|
||||
throw ie;
|
||||
}
|
||||
catch (Exception e) {
|
||||
localLock.unlock();
|
||||
this.localLock.unlock();
|
||||
rethrowAsLockException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean tryLock() {
|
||||
Lock localLock = RedisLockRegistry.this.localRegistry.obtain(this.lockKey);
|
||||
try {
|
||||
if (!localLock.tryLock()) {
|
||||
return false;
|
||||
return tryLock(0, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
|
||||
long now = System.currentTimeMillis();
|
||||
if (!this.localLock.tryLock(time, unit)) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
long expire = now + TimeUnit.MILLISECONDS.convert(time, unit);
|
||||
boolean acquired;
|
||||
while (!(acquired = obtainLock()) && System.currentTimeMillis() < expire) { //NOSONAR
|
||||
Thread.sleep(100); //NOSONAR
|
||||
}
|
||||
boolean obtainedLock = this.obtainLock();
|
||||
if (!obtainedLock) {
|
||||
localLock.unlock();
|
||||
if (!acquired) {
|
||||
this.localLock.unlock();
|
||||
}
|
||||
return obtainedLock;
|
||||
return acquired;
|
||||
}
|
||||
catch (Exception e) {
|
||||
localLock.unlock();
|
||||
this.localLock.unlock();
|
||||
rethrowAsLockException(e);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private boolean obtainLock() {
|
||||
Thread currentThread = Thread.currentThread();
|
||||
if (currentThread.equals(this.thread)) {
|
||||
this.reLock++;
|
||||
return true;
|
||||
boolean success = RedisLockRegistry.this.redisTemplate.execute(RedisLockRegistry.this.obtainLockScript,
|
||||
Collections.singletonList(this.lockKey), RedisLockRegistry.this.clientId,
|
||||
String.valueOf(RedisLockRegistry.this.expireAfter));
|
||||
if (success) {
|
||||
this.lockedAt = System.currentTimeMillis();
|
||||
}
|
||||
|
||||
toHardThreadStorage(this);
|
||||
|
||||
/*
|
||||
* Set these now so they will be persisted if successful.
|
||||
*/
|
||||
this.lockedAt = System.currentTimeMillis();
|
||||
this.threadName = currentThread.getName();
|
||||
|
||||
Boolean success = false;
|
||||
try {
|
||||
|
||||
success = RedisLockRegistry.this.redisTemplate.execute((RedisCallback<Boolean>) connection -> {
|
||||
|
||||
/*
|
||||
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))
|
||||
};
|
||||
|
||||
return connection.execute("SET", actualArgs) != null;
|
||||
});
|
||||
}
|
||||
finally {
|
||||
|
||||
if (!success) {
|
||||
this.lockedAt = 0;
|
||||
this.threadName = null;
|
||||
toWeakThreadStorage(this);
|
||||
}
|
||||
else {
|
||||
this.thread = currentThread;
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("New lock; " + this);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
return success;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
|
||||
Lock localLock = RedisLockRegistry.this.localRegistry.obtain(this.lockKey);
|
||||
if (!localLock.tryLock(time, unit)) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
long expire = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(time, unit);
|
||||
boolean acquired;
|
||||
while (!(acquired = obtainLock()) && System.currentTimeMillis() < expire) { //NOSONAR
|
||||
Thread.sleep(100); //NOSONAR
|
||||
}
|
||||
if (!acquired) {
|
||||
localLock.unlock();
|
||||
}
|
||||
return acquired;
|
||||
}
|
||||
catch (Exception e) {
|
||||
localLock.unlock();
|
||||
rethrowAsLockException(e);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unlock() {
|
||||
if (!Thread.currentThread().equals(this.thread)) {
|
||||
if (this.thread == null) {
|
||||
throw new IllegalStateException("Lock is not locked; " + this);
|
||||
}
|
||||
throw new IllegalStateException("Lock is owned by " + this.thread.getName() + "; " + this);
|
||||
if (!this.localLock.isHeldByCurrentThread()) {
|
||||
throw new IllegalStateException("You do not own lock at " + this.lockKey);
|
||||
}
|
||||
if (this.localLock.getHoldCount() > 1) {
|
||||
this.localLock.unlock();
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
if (this.reLock-- <= 0) {
|
||||
try {
|
||||
this.assertLockInRedisIsUnchanged();
|
||||
RedisLockRegistry.this.redisTemplate.delete(constructLockKey());
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Released lock; " + this);
|
||||
}
|
||||
}
|
||||
finally {
|
||||
this.thread = null;
|
||||
this.reLock = 0;
|
||||
toWeakThreadStorage(this);
|
||||
}
|
||||
RedisLockRegistry.this.redisTemplate.delete(this.lockKey);
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Released lock; " + this);
|
||||
}
|
||||
}
|
||||
finally {
|
||||
Lock localLock = RedisLockRegistry.this.localRegistry.obtain(this.lockKey);
|
||||
localLock.unlock();
|
||||
this.localLock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private void assertLockInRedisIsUnchanged() {
|
||||
RedisLock lockInStore = RedisLockRegistry.this.redisTemplate.boundValueOps(
|
||||
constructLockKey()).get();
|
||||
if (lockInStore == null || !this.equals(lockInStore)) {
|
||||
throw new IllegalStateException("Lock was released due to expiration; " + this
|
||||
+ (lockInStore == null ? "" : "; lock in store: " + lockInStore));
|
||||
}
|
||||
}
|
||||
|
||||
private String constructLockKey() {
|
||||
return RedisLockRegistry.this.registryKey + ":" + this.lockKey;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Condition newCondition() {
|
||||
throw new UnsupportedOperationException("Conditions are not supported");
|
||||
}
|
||||
|
||||
public boolean isAcquiredInThisProcess() {
|
||||
return RedisLockRegistry.this.clientId.equals(
|
||||
RedisLockRegistry.this.redisTemplate.boundValueOps(this.lockKey).get());
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "RedisLock [lockKey=" + constructLockKey()
|
||||
+ ",lockedAt=" + DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(
|
||||
Instant.ofEpochMilli(this.lockedAt)
|
||||
.atZone(ZoneId.systemDefault()))
|
||||
+ ", thread=" + this.threadName
|
||||
+ ", lockHost=" + new String(this.lockHost)
|
||||
SimpleDateFormat dateFormat = new SimpleDateFormat("YYYY-MM-dd@HH:mm:ss.SSS");
|
||||
return "RedisLock [lockKey=" + this.lockKey
|
||||
+ ",lockedAt=" + dateFormat.format(new Date(this.lockedAt))
|
||||
+ ", clientId=" + RedisLockRegistry.this.clientId
|
||||
+ "]";
|
||||
}
|
||||
|
||||
@@ -543,10 +300,9 @@ public final class RedisLockRegistry implements LockRegistry {
|
||||
final int prime = 31;
|
||||
int result = 1;
|
||||
result = prime * result + getOuterType().hashCode();
|
||||
result = prime * result + Arrays.hashCode(this.lockHost);
|
||||
result = prime * result + ((this.lockKey == null) ? 0 : this.lockKey.hashCode());
|
||||
result = prime * result + (int) (this.lockedAt ^ (this.lockedAt >>> 32));
|
||||
result = prime * result + ((this.threadName == null) ? 0 : this.threadName.hashCode());
|
||||
result = prime * result + RedisLockRegistry.this.clientId.hashCode();
|
||||
return result;
|
||||
}
|
||||
|
||||
@@ -565,23 +321,12 @@ public final class RedisLockRegistry implements LockRegistry {
|
||||
if (!getOuterType().equals(other.getOuterType())) {
|
||||
return false;
|
||||
}
|
||||
if (!Arrays.equals(this.lockHost, other.lockHost)) {
|
||||
return false;
|
||||
}
|
||||
if (!this.lockKey.equals(other.lockKey)) {
|
||||
return false;
|
||||
}
|
||||
if (this.lockedAt != other.lockedAt) {
|
||||
return false;
|
||||
}
|
||||
if (this.threadName == null) {
|
||||
if (other.threadName != null) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
else if (!this.threadName.equals(other.threadName)) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -591,51 +336,4 @@ public final class RedisLockRegistry implements LockRegistry {
|
||||
|
||||
}
|
||||
|
||||
private class LockSerializer implements RedisSerializer<RedisLock> {
|
||||
|
||||
LockSerializer() {
|
||||
super();
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] serialize(RedisLock t) throws SerializationException {
|
||||
int hostLength = t.lockHost.length;
|
||||
int keyLength = t.lockKey.length();
|
||||
int threadNameLength = t.threadName.length();
|
||||
byte[] value = new byte[1 + hostLength +
|
||||
1 + keyLength +
|
||||
1 + threadNameLength + 8];
|
||||
ByteBuffer buff = ByteBuffer.wrap(value);
|
||||
buff.put((byte) hostLength)
|
||||
.put(t.lockHost)
|
||||
.put((byte) keyLength)
|
||||
.put(t.lockKey.getBytes())
|
||||
.put((byte) threadNameLength)
|
||||
.put(t.threadName.getBytes())
|
||||
.putLong(t.lockedAt);
|
||||
return value;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RedisLock deserialize(byte[] bytes) throws SerializationException {
|
||||
if (bytes == null) {
|
||||
return null;
|
||||
}
|
||||
ByteBuffer buff = ByteBuffer.wrap(bytes);
|
||||
byte[] host = new byte[buff.get()];
|
||||
buff.get(host);
|
||||
byte[] lockKey = new byte[buff.get()];
|
||||
buff.get(lockKey);
|
||||
byte[] threadName = new byte[buff.get()];
|
||||
buff.get(threadName);
|
||||
long lockedAt = buff.getLong();
|
||||
RedisLock lock = new RedisLock(new String(lockKey));
|
||||
lock.lockedAt = lockedAt;
|
||||
lock.lockHost = host;
|
||||
lock.threadName = new String(threadName);
|
||||
return lock;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,175 @@
|
||||
/*
|
||||
* Copyright 2016-2017 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.integration.redis.leader;
|
||||
|
||||
import static org.hamcrest.Matchers.is;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertThat;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.leader.Context;
|
||||
import org.springframework.integration.leader.DefaultCandidate;
|
||||
import org.springframework.integration.leader.event.LeaderEventPublisher;
|
||||
import org.springframework.integration.redis.rules.RedisAvailable;
|
||||
import org.springframework.integration.redis.rules.RedisAvailableTests;
|
||||
import org.springframework.integration.redis.util.RedisLockRegistry;
|
||||
import org.springframework.integration.support.leader.LockRegistryLeaderInitiator;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
* @author Gary Russell
|
||||
*
|
||||
* @since 4.3.9
|
||||
*/
|
||||
public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
|
||||
|
||||
@Test
|
||||
@RedisAvailable
|
||||
public void testDistributedLeaderElection() throws Exception {
|
||||
CountDownLatch granted = new CountDownLatch(1);
|
||||
CountingPublisher countingPublisher = new CountingPublisher(granted);
|
||||
List<LockRegistryLeaderInitiator> initiators = new ArrayList<>();
|
||||
for (int i = 0; i < 2; i++) {
|
||||
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), "LeaderInitiator");
|
||||
LockRegistryLeaderInitiator initiator =
|
||||
new LockRegistryLeaderInitiator(registry, new DefaultCandidate("foo", "bar"));
|
||||
initiator.setLeaderEventPublisher(countingPublisher);
|
||||
initiators.add(initiator);
|
||||
}
|
||||
|
||||
for (LockRegistryLeaderInitiator initiator : initiators) {
|
||||
initiator.start();
|
||||
}
|
||||
|
||||
assertThat(granted.await(10, TimeUnit.SECONDS), is(true));
|
||||
|
||||
LockRegistryLeaderInitiator initiator1 = countingPublisher.initiator;
|
||||
|
||||
LockRegistryLeaderInitiator initiator2 = null;
|
||||
|
||||
for (LockRegistryLeaderInitiator initiator : initiators) {
|
||||
if (initiator != initiator1) {
|
||||
initiator2 = initiator;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
assertNotNull(initiator2);
|
||||
|
||||
assertThat(initiator1.getContext().isLeader(), is(true));
|
||||
assertThat(initiator2.getContext().isLeader(), is(false));
|
||||
|
||||
final CountDownLatch granted1 = new CountDownLatch(1);
|
||||
final CountDownLatch granted2 = new CountDownLatch(1);
|
||||
CountDownLatch revoked1 = new CountDownLatch(1);
|
||||
CountDownLatch revoked2 = new CountDownLatch(1);
|
||||
initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1) {
|
||||
|
||||
@Override
|
||||
public void publishOnRevoked(Object source, Context context, String role) {
|
||||
try {
|
||||
// It's difficult to see round-robin election, so block one initiator until the second is elected.
|
||||
assertThat(granted2.await(10, TimeUnit.SECONDS), is(true));
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
// No op
|
||||
}
|
||||
super.publishOnRevoked(source, context, role);
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
initiator2.setLeaderEventPublisher(new CountingPublisher(granted2, revoked2) {
|
||||
|
||||
@Override
|
||||
public void publishOnRevoked(Object source, Context context, String role) {
|
||||
try {
|
||||
// It's difficult to see round-robin election, so block one initiator until the second is elected.
|
||||
assertThat(granted1.await(10, TimeUnit.SECONDS), is(true));
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
// No op
|
||||
}
|
||||
super.publishOnRevoked(source, context, role);
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
initiator1.getContext().yield();
|
||||
|
||||
assertThat(revoked1.await(10, TimeUnit.SECONDS), is(true));
|
||||
|
||||
assertThat(initiator2.getContext().isLeader(), is(true));
|
||||
assertThat(initiator1.getContext().isLeader(), is(false));
|
||||
|
||||
initiator2.getContext().yield();
|
||||
|
||||
assertThat(revoked2.await(10, TimeUnit.SECONDS), is(true));
|
||||
|
||||
assertThat(initiator1.getContext().isLeader(), is(true));
|
||||
assertThat(initiator2.getContext().isLeader(), is(false));
|
||||
|
||||
initiator2.stop();
|
||||
|
||||
CountDownLatch revoked11 = new CountDownLatch(1);
|
||||
initiator1.setLeaderEventPublisher(new CountingPublisher(new CountDownLatch(1), revoked11));
|
||||
|
||||
initiator1.getContext().yield();
|
||||
|
||||
assertThat(revoked11.await(10, TimeUnit.SECONDS), is(true));
|
||||
assertThat(initiator1.getContext().isLeader(), is(false));
|
||||
|
||||
initiator1.stop();
|
||||
}
|
||||
|
||||
private static class CountingPublisher implements LeaderEventPublisher {
|
||||
|
||||
private CountDownLatch granted;
|
||||
|
||||
private CountDownLatch revoked;
|
||||
|
||||
private volatile LockRegistryLeaderInitiator initiator;
|
||||
|
||||
CountingPublisher(CountDownLatch granted, CountDownLatch revoked) {
|
||||
this.granted = granted;
|
||||
this.revoked = revoked;
|
||||
}
|
||||
|
||||
CountingPublisher(CountDownLatch granted) {
|
||||
this(granted, new CountDownLatch(1));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void publishOnRevoked(Object source, Context context, String role) {
|
||||
this.revoked.countDown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void publishOnGranted(Object source, Context context, String role) {
|
||||
this.initiator = (LockRegistryLeaderInitiator) source;
|
||||
this.granted.countDown();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2014-2016 the original author or authors.
|
||||
* Copyright 2014-2017 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.
|
||||
@@ -21,15 +21,12 @@ import static org.hamcrest.Matchers.instanceOf;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNotEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNotSame;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertSame;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.Executors;
|
||||
@@ -45,10 +42,10 @@ import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.ExpectedException;
|
||||
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.integration.redis.rules.RedisAvailable;
|
||||
import org.springframework.integration.redis.rules.RedisAvailableTests;
|
||||
import org.springframework.integration.test.rule.Log4jLevelAdjuster;
|
||||
@@ -58,6 +55,7 @@ import org.springframework.integration.test.util.TestUtils;
|
||||
* @author Gary Russell
|
||||
* @author Konstantin Yakimov
|
||||
* @author Artem Bilan
|
||||
* @author Vedran Pavic
|
||||
* @since 4.0
|
||||
*
|
||||
*/
|
||||
@@ -72,20 +70,19 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
@Rule
|
||||
public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE, "org.springframework.integration.redis");
|
||||
|
||||
@Rule
|
||||
public ExpectedException thrown = ExpectedException.none();
|
||||
|
||||
@Before
|
||||
@After
|
||||
public void setupShutDown() {
|
||||
RedisTemplate<String, ?> template = this.createTemplate();
|
||||
StringRedisTemplate template = this.createTemplate();
|
||||
template.delete(this.registryKey + ":*");
|
||||
template.delete(this.registryKey2 + ":*");
|
||||
}
|
||||
|
||||
private RedisTemplate<String, ?> createTemplate() {
|
||||
RedisTemplate<String, ?> template = new RedisTemplate<>();
|
||||
template.setConnectionFactory(this.getConnectionFactoryForTest());
|
||||
template.setKeySerializer(new StringRedisSerializer());
|
||||
template.afterPropertiesSet();
|
||||
return template;
|
||||
private StringRedisTemplate createTemplate() {
|
||||
return new StringRedisTemplate(this.getConnectionFactoryForTest());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -96,13 +93,14 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
Lock lock = registry.obtain("foo");
|
||||
lock.lock();
|
||||
try {
|
||||
assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
registry.expireUnusedOlderThan(-1000);
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -113,13 +111,14 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
Lock lock = registry.obtain("foo");
|
||||
lock.lockInterruptibly();
|
||||
try {
|
||||
assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
registry.expireUnusedOlderThan(-1000);
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -144,7 +143,8 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
lock1.unlock();
|
||||
}
|
||||
}
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
registry.expireUnusedOlderThan(-1000);
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -169,7 +169,8 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
lock1.unlock();
|
||||
}
|
||||
}
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
registry.expireUnusedOlderThan(-1000);
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -194,7 +195,8 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
lock1.unlock();
|
||||
}
|
||||
}
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
registry.expireUnusedOlderThan(-1000);
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -222,8 +224,9 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
lock1.unlock();
|
||||
Object ise = result.get(10, TimeUnit.SECONDS);
|
||||
assertThat(ise, instanceOf(IllegalStateException.class));
|
||||
assertThat(((Exception) ise).getMessage(), containsString("Lock is not locked"));
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertThat(((Exception) ise).getMessage(), containsString("You do not own lock at"));
|
||||
registry.expireUnusedOlderThan(-1000);
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -236,13 +239,13 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
final CountDownLatch latch2 = new CountDownLatch(1);
|
||||
final CountDownLatch latch3 = new CountDownLatch(1);
|
||||
lock1.lockInterruptibly();
|
||||
assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
Executors.newSingleThreadExecutor().execute(() -> {
|
||||
Lock lock2 = registry.obtain("foo");
|
||||
try {
|
||||
latch1.countDown();
|
||||
lock2.lockInterruptibly();
|
||||
assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
latch2.await(10, TimeUnit.SECONDS);
|
||||
locked.set(true);
|
||||
}
|
||||
@@ -260,7 +263,8 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
latch2.countDown();
|
||||
assertTrue(latch3.await(10, TimeUnit.SECONDS));
|
||||
assertTrue(locked.get());
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
registry.expireUnusedOlderThan(-1000);
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -274,19 +278,19 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
final CountDownLatch latch2 = new CountDownLatch(1);
|
||||
final CountDownLatch latch3 = new CountDownLatch(1);
|
||||
lock1.lockInterruptibly();
|
||||
assertNotNull(TestUtils.getPropertyValue(registry1, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertEquals(1, TestUtils.getPropertyValue(registry1, "locks", Map.class).size());
|
||||
Executors.newSingleThreadExecutor().execute(() -> {
|
||||
Lock lock2 = registry2.obtain("foo");
|
||||
try {
|
||||
latch1.countDown();
|
||||
lock2.lockInterruptibly();
|
||||
assertNotNull(TestUtils.getPropertyValue(registry2, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertEquals(1, TestUtils.getPropertyValue(registry2, "locks", Map.class).size());
|
||||
latch2.await(10, TimeUnit.SECONDS);
|
||||
locked.set(true);
|
||||
}
|
||||
catch (InterruptedException e1) {
|
||||
Thread.currentThread().interrupt();
|
||||
logger.error("Interrupted while locking: " + lock2, e1);
|
||||
this.logger.error("Interrupted while locking: " + lock2, e1);
|
||||
}
|
||||
finally {
|
||||
try {
|
||||
@@ -294,7 +298,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
latch3.countDown();
|
||||
}
|
||||
catch (IllegalStateException e2) {
|
||||
logger.error("Failed to unlock: " + lock2, e2);
|
||||
this.logger.error("Failed to unlock: " + lock2, e2);
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -304,8 +308,10 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
latch2.countDown();
|
||||
assertTrue(latch3.await(10, TimeUnit.SECONDS));
|
||||
assertTrue(locked.get());
|
||||
assertNull(TestUtils.getPropertyValue(registry1, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertNull(TestUtils.getPropertyValue(registry2, "hardThreadLocks", ThreadLocal.class).get());
|
||||
registry1.expireUnusedOlderThan(-1000);
|
||||
registry2.expireUnusedOlderThan(-1000);
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry1, "locks", Map.class).size());
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry2, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -331,84 +337,23 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
lock.unlock();
|
||||
Object ise = result.get(10, TimeUnit.SECONDS);
|
||||
assertThat(ise, instanceOf(IllegalStateException.class));
|
||||
assertThat(((Exception) ise).getMessage(), containsString("Lock is owned by"));
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertThat(((Exception) ise).getMessage(), containsString("You do not own lock at"));
|
||||
registry.expireUnusedOlderThan(-1000);
|
||||
assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@RedisAvailable
|
||||
public void testList() throws Exception {
|
||||
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
|
||||
Lock foo = registry.obtain("foo");
|
||||
foo.lockInterruptibly();
|
||||
Lock bar = registry.obtain("bar");
|
||||
bar.lockInterruptibly();
|
||||
Lock baz = registry.obtain("baz");
|
||||
baz.lockInterruptibly();
|
||||
Collection<Lock> locks = registry.listLocks();
|
||||
assertEquals(3, locks.size());
|
||||
foo.unlock();
|
||||
bar.unlock();
|
||||
baz.unlock();
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
}
|
||||
|
||||
@Test
|
||||
@RedisAvailable
|
||||
public void testExpireNoLockInStore() throws Exception {
|
||||
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey, 100);
|
||||
Lock foo = registry.obtain("foo");
|
||||
foo.lockInterruptibly();
|
||||
public void testExpireTwoRegistries() throws Exception {
|
||||
RedisLockRegistry registry1 = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey, 100);
|
||||
RedisLockRegistry registry2 = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey, 100);
|
||||
Lock lock1 = registry1.obtain("foo");
|
||||
Lock lock2 = registry2.obtain("foo");
|
||||
assertTrue(lock1.tryLock());
|
||||
assertFalse(lock2.tryLock());
|
||||
waitForExpire("foo");
|
||||
try {
|
||||
foo.unlock();
|
||||
fail("Expected exception");
|
||||
}
|
||||
catch (IllegalStateException e) {
|
||||
assertThat(e.getMessage(), containsString("Lock was released due to expiration"));
|
||||
}
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
}
|
||||
|
||||
@Test
|
||||
@RedisAvailable
|
||||
public void testExpireDuringSecondObtain() throws Exception {
|
||||
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey, 100);
|
||||
registry.setUseWeakReferences(true);
|
||||
Lock foo = registry.obtain("foo");
|
||||
foo.lockInterruptibly();
|
||||
waitForExpire("foo");
|
||||
Lock foo1 = registry.obtain("foo");
|
||||
assertNotSame(foo, foo1);
|
||||
|
||||
try {
|
||||
foo.unlock();
|
||||
fail("IllegalStateException");
|
||||
}
|
||||
catch (IllegalStateException e) {
|
||||
assertThat(e.getMessage(), containsString("Lock is not locked"));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@RedisAvailable
|
||||
public void testExpireNewLockInStore() throws Exception {
|
||||
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey, 100);
|
||||
Lock foo1 = registry.obtain("foo");
|
||||
foo1.lockInterruptibly();
|
||||
waitForExpire("foo");
|
||||
Lock foo2 = registry.obtain("foo");
|
||||
assertNotSame(foo1, foo2);
|
||||
foo2.lockInterruptibly();
|
||||
try {
|
||||
foo1.unlock();
|
||||
fail("Expected exception");
|
||||
}
|
||||
catch (IllegalStateException e) {
|
||||
assertThat(e.getMessage(), containsString("Lock is not locked"));
|
||||
}
|
||||
foo2.unlock();
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertTrue(lock2.tryLock());
|
||||
assertFalse(lock1.tryLock());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -416,7 +361,6 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
public void testEquals() throws Exception {
|
||||
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
|
||||
RedisLockRegistry registry1 = new RedisLockRegistry(connectionFactory, this.registryKey);
|
||||
registry1.setUseWeakReferences(true);
|
||||
RedisLockRegistry registry2 = new RedisLockRegistry(connectionFactory, this.registryKey);
|
||||
RedisLockRegistry registry3 = new RedisLockRegistry(connectionFactory, this.registryKey2);
|
||||
Lock lock1 = registry1.obtain("foo");
|
||||
@@ -449,27 +393,23 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
@RedisAvailable
|
||||
public void testThreadLocalListLeaks() {
|
||||
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey, 100);
|
||||
registry.setUseWeakReferences(true);
|
||||
|
||||
for (int i = 0; i < 10; i++) {
|
||||
registry.obtain("foo" + i);
|
||||
}
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertEquals(10, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
|
||||
for (int i = 0; i < 10; i++) {
|
||||
Lock lock = registry.obtain("foo" + i);
|
||||
lock.lock();
|
||||
}
|
||||
assertEquals(10,
|
||||
((Collection<?>) TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()).size());
|
||||
assertNull(TestUtils.getPropertyValue(registry, "weakThreadLocks", ThreadLocal.class).get());
|
||||
assertEquals(10, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
|
||||
for (int i = 0; i < 10; i++) {
|
||||
Lock lock = registry.obtain("foo" + i);
|
||||
assertNotNull(TestUtils.getPropertyValue(lock, "thread", Thread.class));
|
||||
lock.unlock();
|
||||
}
|
||||
assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
|
||||
assertEquals(10, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -493,13 +433,13 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
|
||||
}
|
||||
|
||||
private Long getExpire(RedisLockRegistry registry, String lockKey) {
|
||||
RedisTemplate<String, ?> template = this.createTemplate();
|
||||
StringRedisTemplate template = this.createTemplate();
|
||||
String registryKey = TestUtils.getPropertyValue(registry, "registryKey", String.class);
|
||||
return template.getExpire(registryKey + ":" + lockKey);
|
||||
}
|
||||
|
||||
private void waitForExpire(String key) throws Exception {
|
||||
RedisTemplate<String, ?> template = this.createTemplate();
|
||||
StringRedisTemplate template = this.createTemplate();
|
||||
int n = 0;
|
||||
while (n++ < 100 && template.keys(this.registryKey + ":" + key).size() > 0) {
|
||||
Thread.sleep(100);
|
||||
|
||||
@@ -764,3 +764,5 @@ Locks are normally held for a much smaller time.
|
||||
IMPORTANT: Because the keys can expire, an attempt to unlock an expired lock will result in an exception being thrown.
|
||||
However, be aware that the resources protected by such a lock may have been compromised so such exceptions should be considered severe.
|
||||
The expiry should be set at a large enough value to prevent this condition, while small enough that the lock can be recovered after a server failure in a reasonable amount of time.
|
||||
|
||||
Starting with _version 5.0_, the `RedisLockRegistry` implements `ExpirableLockRegistry` providing functionality to remove locks last acquired more than `age` ago that are not currently locked.
|
||||
|
||||
Reference in New Issue
Block a user