From 3c32daa25a7077f31049089cb56fd044807b1a68 Mon Sep 17 00:00:00 2001 From: unseok kim Date: Sun, 12 Dec 2021 22:06:17 -0500 Subject: [PATCH] GH-3690: Use pub-sub in RedisLockReg for unlocks Fixes https://github.com/spring-projects/spring-integration/issues/3690 Replace spinLock with pub-sub rename and add comment * change remove timing for 'notifyMap' * use unlink * remove DUMMY_CALLABLE * Clean up code style --- .../redis/util/RedisLockRegistry.java | 168 +++++++++++++++--- .../redis/util/RedisLockRegistryTests.java | 87 +++++++++ 2 files changed, 227 insertions(+), 28 deletions(-) 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 719f387c97..2208a90579 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 @@ -23,10 +23,14 @@ import java.util.LinkedHashMap; import java.util.Map; import java.util.Map.Entry; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -36,14 +40,20 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; import org.springframework.dao.CannotAcquireLockException; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.RedisConnectionFactory; 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.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.data.redis.listener.Topic; import org.springframework.integration.support.locks.ExpirableLockRegistry; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; +import org.springframework.util.concurrent.SettableListenableFuture; /** * Implementation of {@link ExpirableLockRegistry} providing a distributed lock using Redis. @@ -93,6 +103,20 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl "end\n" + "return false"; + private static final String UNLINK_UNLOCK_SCRIPT = + "if (redis.call('unlink', KEYS[1]) == 1) then " + + "redis.call('publish', ARGV[1], KEYS[1]) " + + "return true " + + "end " + + "return false"; + + private static final String DELETE_UNLOCK_SCRIPT = + "if (redis.call('del', KEYS[1]) == 1) then " + + "redis.call('publish', ARGV[1], KEYS[1]) " + + "return true " + + "end " + + "return false"; + private final Map locks = new LinkedHashMap(16, 0.75F, true) { @@ -108,10 +132,20 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl private final String registryKey; + private final String unLockChannelKey; + private final StringRedisTemplate redisTemplate; private final RedisScript obtainLockScript; + private final RedisScript unLinkUnLockScript; + + private final RedisScript deleteUnLockScript; + + private final RedisUnLockNotifyMessageListener unlockNotifyMessageListener; + + private final RedisMessageListenerContainer redisMessageListenerContainer; + private final long expireAfter; private int cacheCapacity = DEFAULT_CAPACITY; @@ -151,8 +185,22 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl Assert.notNull(registryKey, "'registryKey' cannot be null"); this.redisTemplate = new StringRedisTemplate(connectionFactory); this.obtainLockScript = new DefaultRedisScript<>(OBTAIN_LOCK_SCRIPT, Boolean.class); + this.unLinkUnLockScript = new DefaultRedisScript<>(UNLINK_UNLOCK_SCRIPT, Boolean.class); + this.deleteUnLockScript = new DefaultRedisScript<>(DELETE_UNLOCK_SCRIPT, Boolean.class); this.registryKey = registryKey; this.expireAfter = expireAfter; + this.unLockChannelKey = registryKey + "-channel"; + this.unlockNotifyMessageListener = new RedisUnLockNotifyMessageListener(); + this.redisMessageListenerContainer = new RedisMessageListenerContainer(); + setupUnlockMessageListener(connectionFactory); + } + + private void setupUnlockMessageListener(RedisConnectionFactory connectionFactory) { + final Topic topic = new ChannelTopic(this.unLockChannelKey); + this.redisMessageListenerContainer.setConnectionFactory(connectionFactory); + this.redisMessageListenerContainer.setTaskExecutor(this.executor); + this.redisMessageListenerContainer.setSubscriptionExecutor(this.executor); + this.redisMessageListenerContainer.addMessageListener(this.unlockNotifyMessageListener, topic); } /** @@ -164,6 +212,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl public void setExecutor(Executor executor) { this.executor = executor; this.executorExplicitlySet = true; + this.redisMessageListenerContainer.setTaskExecutor(this.executor); + this.redisMessageListenerContainer.setSubscriptionExecutor(this.executor); } /** @@ -189,7 +239,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl long now = System.currentTimeMillis(); synchronized (this.locks) { this.locks.entrySet() - .removeIf((entry) -> { + .removeIf(entry -> { RedisLock lock = entry.getValue(); return now - lock.getLockedAt() > age && !lock.isAcquiredInThisProcess(); }); @@ -201,6 +251,12 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl if (!this.executorExplicitlySet) { ((ExecutorService) this.executor).shutdown(); } + try { + this.redisMessageListenerContainer.destroy(); + } + catch (Exception ex) { + throw new IllegalStateException(ex); + } } private final class RedisLock implements Lock { @@ -228,8 +284,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl this.localLock.lock(); while (true) { try { - while (!obtainLock()) { - Thread.sleep(100); //NOSONAR + while (!subscribeLock()) { + // empty } break; } @@ -255,8 +311,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl public void lockInterruptibly() throws InterruptedException { this.localLock.lockInterruptibly(); try { - while (!obtainLock()) { - Thread.sleep(100); //NOSONAR + while (!subscribeLock()) { + // empty } } catch (InterruptedException ie) { @@ -283,16 +339,12 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl @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 - } + long waitTime = TimeUnit.MILLISECONDS.convert(time, unit); + boolean acquired = subscribeLock(waitTime); if (!acquired) { this.localLock.unlock(); } @@ -305,6 +357,38 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl return false; } + private boolean subscribeLock() throws ExecutionException, InterruptedException { + return subscribeLock(-1L); + } + + private boolean subscribeLock(long time) throws ExecutionException, InterruptedException { + if (!obtainLock()) { + if (!RedisLockRegistry.this.redisMessageListenerContainer.isRunning()) { + RedisLockRegistry.this.redisMessageListenerContainer.afterPropertiesSet(); + RedisLockRegistry.this.redisMessageListenerContainer.start(); + } + try { + Future future = + RedisLockRegistry.this.unlockNotifyMessageListener.subscribeLock(this.lockKey); + //DCL + if (!obtainLock()) { + try { + //if short expireAfter key expire for ttl, no receive unlock msg + long waitTime = time >= 0 ? time : RedisLockRegistry.this.expireAfter; + future.get(waitTime, TimeUnit.MILLISECONDS); + } + catch (TimeoutException ignore) { + } + return obtainLock(); + } + } + finally { + RedisLockRegistry.this.unlockNotifyMessageListener.unSubscribeLock(this.lockKey); + } + } + return true; + } + private boolean obtainLock() { Boolean success = RedisLockRegistry.this.redisTemplate.execute(RedisLockRegistry.this.obtainLockScript, @@ -354,25 +438,26 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl } private void removeLockKey() { - if (RedisLockRegistry.this.unlinkAvailable) { - try { - RedisLockRegistry.this.redisTemplate.unlink(this.lockKey); - } - catch (Exception ex) { - RedisLockRegistry.this.unlinkAvailable = false; - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("The UNLINK command has failed (not supported on the Redis server?); " + - "falling back to the regular DELETE command", ex); - } - else { - LOGGER.warn("The UNLINK command has failed (not supported on the Redis server?); " + - "falling back to the regular DELETE command: " + ex.getMessage()); - } - RedisLockRegistry.this.redisTemplate.delete(this.lockKey); + try { + if (RedisLockRegistry.this.unlinkAvailable) { + RedisLockRegistry.this.redisTemplate.execute( + RedisLockRegistry.this.unLinkUnLockScript, Collections.singletonList(this.lockKey), + RedisLockRegistry.this.unLockChannelKey); } } - else { - RedisLockRegistry.this.redisTemplate.delete(this.lockKey); + catch (Exception ex) { + RedisLockRegistry.this.unlinkAvailable = false; + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("The UNLINK command has failed (not supported on the Redis server?); " + + "falling back to the regular DELETE command", ex); + } + else { + LOGGER.warn("The UNLINK command has failed (not supported on the Redis server?); " + + "falling back to the regular DELETE command: " + ex.getMessage()); + } + RedisLockRegistry.this.redisTemplate.execute( + RedisLockRegistry.this.deleteUnLockScript, Collections.singletonList(this.lockKey), + RedisLockRegistry.this.unLockChannelKey); } } @@ -433,4 +518,31 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl } + private static final class RedisUnLockNotifyMessageListener implements MessageListener { + + private final Map> notifyMap = new ConcurrentHashMap<>(); + + @Override + public void onMessage(Message message, byte[] pattern) { + final String lockKey = new String(message.getBody()); + unlockNotify(lockKey); + } + + public Future subscribeLock(String lockKey) { + return this.notifyMap.computeIfAbsent(lockKey, key -> new SettableListenableFuture<>()); + } + + public void unSubscribeLock(String localLock) { + this.notifyMap.remove(localLock); + } + + private void unlockNotify(String lockKey) { + this.notifyMap.computeIfPresent(lockKey, (key, lockFuture) -> { + lockFuture.set(key); + return lockFuture; + }); + } + + } + } 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 dfe3d25859..f6c473f7eb 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 @@ -22,18 +22,25 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.BDDMockito.willReturn; import static org.mockito.Mockito.mock; +import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Queue; import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Lock; +import java.util.stream.Collectors; +import java.util.stream.IntStream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -606,6 +613,85 @@ public class RedisLockRegistryTests extends RedisAvailableTests { assertThat(getRedisLockRegistryLocks(registry)).containsKeys("foo:3", "foo:4", "foo:5"); } + @RedisAvailable + @Test + public void twoRedisLockRegistryTest() throws InterruptedException { + RedisConnectionFactory connectionFactory = getConnectionFactoryForTest(); + RedisLockRegistry registry1 = new RedisLockRegistry(connectionFactory, registryKey, 1000000L); + RedisLockRegistry registry2 = new RedisLockRegistry(connectionFactory, registryKey, 1000000L); + String lockKey = "test-1"; + + Lock obtainLock_1 = registry1.obtain(lockKey); + Lock obtainLock_2 = registry2.obtain(lockKey); + + CountDownLatch registry1Lock = new CountDownLatch(1); + CountDownLatch endDownLatch = new CountDownLatch(2); + + CompletableFuture.runAsync(() -> { + try { + obtainLock_1.lock(); + // for (int i = 0; i < 10; i++) { + // Thread.sleep(1000); + // } + registry1Lock.countDown(); + obtainLock_1.unlock(); + endDownLatch.countDown(); + } + catch (Exception ignore) { + ignore.printStackTrace(); + } + }); + + CompletableFuture.runAsync(() -> { + try { + registry1Lock.await(); + } + catch (InterruptedException ignore) { + } + obtainLock_2.lock(); + obtainLock_2.unlock(); + endDownLatch.countDown(); + }); + + endDownLatch.await(); + } + + @RedisAvailable + @Test + public void multiRedisLockRegistryTest() throws InterruptedException, ExecutionException { + final RedisConnectionFactory connectionFactory = getConnectionFactoryForTest(); + final String testKey = "testKey"; + final long expireAfter = 100000L; + final int lockRegistryNum = 10; + final ExecutorService executorService = Executors.newFixedThreadPool(lockRegistryNum * 2); + final AtomicInteger atomicInteger = new AtomicInteger(0); + final List> collect = IntStream.range(0, lockRegistryNum) + .mapToObj((num) -> new RedisLockRegistry( + connectionFactory, registryKey, expireAfter)) + .map((registry) -> { + final Callable callable = () -> { + Lock obtain = registry.obtain(testKey); + obtain.lock(); + obtain.unlock(); + atomicInteger.incrementAndGet(); + return true; + }; + return callable; + }) + .collect(Collectors.toList()); + + final int testCnt = 3; + for (int i = 0; i < testCnt; i++) { + List> futures_1 = executorService.invokeAll(collect); + for (Future fu : futures_1) { + fu.get(); + } + } + + assertThat(atomicInteger.get()).isEqualTo(testCnt * lockRegistryNum); + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testUlink() { @@ -639,4 +725,5 @@ public class RedisLockRegistryTests extends RedisAvailableTests { private static Map getRedisLockRegistryLocks(RedisLockRegistry registry) { return TestUtils.getPropertyValue(registry, "locks", Map.class); } + }