diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java index f355b67649..f0b0032ed7 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java @@ -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; *

* Note: This is not intended for low latency applications. 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. - *

- * 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. *

* {@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 locks = new ConcurrentHashMap<>(); + + private final String clientId = UUID.randomUUID().toString(); + private final String registryKey; - private final RedisTemplate redisTemplate; + private final StringRedisTemplate redisTemplate; - private final ThreadLocal> weakThreadLocks = new ThreadLocal>(); - - private final ThreadLocal> hardThreadLocks = new ThreadLocal>(); + private final RedisScript 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). - *

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 getWeakThreadLocks() { - Set locks = this.weakThreadLocks.get(); - if (locks == null) { - locks = Collections.newSetFromMap(new WeakHashMap()); - this.weakThreadLocks.set(locks); - } - return locks; - } - - /** - * Hard referenced locks, lock is kept here when actual lock is gained. - */ - private Collection getHardThreadLocks() { - List locks = this.hardThreadLocks.get(); - if (locks == null) { - locks = new LinkedList<>(); - this.hardThreadLocks.set(locks); - } - return locks; - } - - private RedisLock findLock(Collection 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 listLocks() { - return this.redisTemplate.execute((RedisCallback>) connection -> { - Set keys = connection.keys((RedisLockRegistry.this.registryKey + ":*").getBytes()); - if (keys.size() > 0) { - List 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> iterator = this.locks.entrySet().iterator(); + long now = System.currentTimeMillis(); + while (iterator.hasNext()) { + Map.Entry 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) 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 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 { - - 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; - } - - } - } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java new file mode 100644 index 0000000000..ec0ffc5e1b --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java @@ -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 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(); + } + + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java index cc156a93e8..ef3df49eef 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java @@ -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 template = this.createTemplate(); + StringRedisTemplate template = this.createTemplate(); template.delete(this.registryKey + ":*"); template.delete(this.registryKey2 + ":*"); } - private RedisTemplate createTemplate() { - RedisTemplate 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 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 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 template = this.createTemplate(); + StringRedisTemplate template = this.createTemplate(); int n = 0; while (n++ < 100 && template.keys(this.registryKey + ":" + key).size() > 0) { Thread.sleep(100); diff --git a/src/reference/asciidoc/redis.adoc b/src/reference/asciidoc/redis.adoc index 933b17ce8e..97b2140062 100644 --- a/src/reference/asciidoc/redis.adoc +++ b/src/reference/asciidoc/redis.adoc @@ -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.