diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java index 969f3c3ade..db3fa4b556 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2024 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. @@ -16,12 +16,15 @@ package org.springframework.integration.support.locks; +import org.springframework.scheduling.TaskScheduler; + /** * A {@link LockRegistry} implementing this interface supports the renewal * of the time to live of a lock. * * @author Alexandre Strubel * @author Artem Bilan + * @author Youbin Wu * * @since 5.4 */ @@ -34,4 +37,14 @@ public interface RenewableLockRegistry extends LockRegistry { */ void renewLock(Object lockKey); + /** + * Set the {@link TaskScheduler} to use for the renewal task. + * When renewalTaskScheduler is set, it will be used to periodically renew the lock to ensure that + * the lock does not expire while the thread is working. + * @param renewalTaskScheduler renew task scheduler + * @since 6.4 + */ + default void setRenewalTaskScheduler(TaskScheduler renewalTaskScheduler) { + } + } 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 5b05127fa7..6796814150 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,6 +16,7 @@ package org.springframework.integration.redis.util; +import java.io.Serial; import java.text.SimpleDateFormat; import java.time.Duration; import java.util.Collections; @@ -32,6 +33,7 @@ import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.locks.Condition; @@ -54,6 +56,8 @@ 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.integration.support.locks.RenewableLockRegistry; +import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; @@ -89,11 +93,12 @@ import org.springframework.util.ReflectionUtils; * @author Myeonghyeon Lee * @author Roman Zabaluev * @author Alex Peelman + * @author Youbin Wu * * @since 4.0 * */ -public final class RedisLockRegistry implements ExpirableLockRegistry, DisposableBean { +public final class RedisLockRegistry implements ExpirableLockRegistry, DisposableBean, RenewableLockRegistry { private static final Log LOGGER = LogFactory.getLog(RedisLockRegistry.class); @@ -110,6 +115,9 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl private final Map locks = new LinkedHashMap<>(16, 0.75F, true) { + @Serial + private static final long serialVersionUID = 7419938441348450459L; + @Override protected boolean removeEldestEntry(Entry eldest) { return size() > RedisLockRegistry.this.cacheCapacity; @@ -138,6 +146,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl private Executor executor = Executors.newCachedThreadPool(new CustomizableThreadFactory("redis-lock-registry-")); + private TaskScheduler renewalTaskScheduler; + /** * Flag to denote whether the {@link ExecutorService} was provided via the setter and * thus should not be shutdown when {@link #destroy()} is called @@ -207,6 +217,12 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl this.executorExplicitlySet = true; } + @Override + public void setRenewalTaskScheduler(TaskScheduler renewalTaskScheduler) { + Assert.notNull(renewalTaskScheduler, "'renewalTaskScheduler' must not be null"); + this.renewalTaskScheduler = renewalTaskScheduler; + } + /** * Set the capacity of cached locks. * @param cacheCapacity The capacity of cached lock, (default 100_000). @@ -291,6 +307,26 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl } } + @Override + public void renewLock(Object lockKey) { + String path = (String) lockKey; + RedisLock redisLock; + this.lock.lock(); + try { + redisLock = this.locks.computeIfAbsent(path, getRedisLockConstructor(this.redisLockType)); + } + finally { + this.lock.unlock(); + } + if (redisLock == null) { + throw new IllegalStateException("Could not renew mutex at " + path); + } + + if (!redisLock.renew()) { + throw new IllegalStateException("Could not renew mutex at " + path); + } + } + /** * The mode in which this registry is going to work with locks. */ @@ -328,8 +364,19 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl return false """; - protected static final RedisScript - OBTAIN_LOCK_REDIS_SCRIPT = new DefaultRedisScript<>(OBTAIN_LOCK_SCRIPT, Boolean.class); + private static final String RENEW_SCRIPT = """ + if (redis.call('GET', KEYS[1]) == ARGV[1]) then + redis.call('PEXPIRE', KEYS[1], ARGV[2]) + return true + end + return false + """; + + protected static final RedisScript OBTAIN_LOCK_REDIS_SCRIPT = + new DefaultRedisScript<>(OBTAIN_LOCK_SCRIPT, Boolean.class); + + public static final RedisScript RENEW_REDIS_SCRIPT = + new DefaultRedisScript<>(RENEW_SCRIPT, Boolean.class); protected final String lockKey; @@ -337,6 +384,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl private volatile long lockedAt; + private volatile ScheduledFuture renewFuture; + private RedisLock(String path) { this.lockKey = constructLockKey(path); } @@ -454,6 +503,11 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl LOGGER.debug("Acquired lock; " + this); } this.lockedAt = System.currentTimeMillis(); + if (RedisLockRegistry.this.renewalTaskScheduler != null) { + Duration delay = Duration.ofMillis(RedisLockRegistry.this.expireAfter / 3); + this.renewFuture = + RedisLockRegistry.this.renewalTaskScheduler.scheduleWithFixedDelay(this::renew, delay); + } } return acquired; } @@ -515,6 +569,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl if (Boolean.TRUE.equals(unlinkResult)) { // Lock key successfully unlinked + stopRenew(); return; } else if (Boolean.FALSE.equals(unlinkResult)) { @@ -526,6 +581,26 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl throw new ConcurrentModificationException("Lock was released in the store due to expiration. " + "The integrity of data protected by this lock may have been compromised."); } + else { + stopRenew(); + } + } + + protected final boolean renew() { + boolean res = Boolean.TRUE.equals(RedisLockRegistry.this.redisTemplate.execute( + RENEW_REDIS_SCRIPT, Collections.singletonList(this.lockKey), + RedisLockRegistry.this.clientId, String.valueOf(RedisLockRegistry.this.expireAfter))); + if (!res) { + stopRenew(); + } + return res; + } + + protected final void stopRenew() { + if (this.renewFuture != null) { + this.renewFuture.cancel(true); + this.renewFuture = null; + } } @Override @@ -553,7 +628,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl int result = 1; result = prime * result + getOuterType().hashCode(); result = prime * result + ((this.lockKey == null) ? 0 : this.lockKey.hashCode()); - result = prime * result + (int) (this.lockedAt ^ (this.lockedAt >>> 32)); // NOSONAR magic number + result = prime * result + Long.hashCode(this.lockedAt); result = prime * result + RedisLockRegistry.this.clientId.hashCode(); return result; } 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 15e9937dd5..bb0313a1f1 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 @@ -51,8 +51,10 @@ import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.integration.redis.RedisContainerTest; import org.springframework.integration.redis.util.RedisLockRegistry.RedisLockType; import org.springframework.integration.test.util.TestUtils; +import org.springframework.scheduling.concurrent.SimpleAsyncTaskScheduler; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.mock; @@ -66,6 +68,7 @@ import static org.mockito.Mockito.mock; * @author Artem Vozhdayenko * @author Anton Gabov * @author Eddie Cho + * @author Youbin Wu * * @since 4.0 * @@ -427,6 +430,20 @@ class RedisLockRegistryTests implements RedisContainerTest { registry.destroy(); } + @ParameterizedTest + @EnumSource(RedisLockType.class) + void testRenewalOnExpire(RedisLockType redisLockType) throws Exception { + long expireAfter = 300L; + RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey, expireAfter); + registry.setRenewalTaskScheduler(new SimpleAsyncTaskScheduler()); + registry.setRedisLockType(redisLockType); + Lock lock1 = registry.obtain("foo"); + assertThat(lock1.tryLock()).isTrue(); + Thread.sleep(expireAfter * 2); + lock1.unlock(); + registry.destroy(); + } + @ParameterizedTest @EnumSource(RedisLockType.class) void testEquals(RedisLockType testRedisLockType) { @@ -900,6 +917,33 @@ class RedisLockRegistryTests implements RedisContainerTest { registry.destroy(); } + @ParameterizedTest + @EnumSource(RedisLockType.class) + void testLockRenew(RedisLockType redisLockType) { + final RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey); + registry.setRedisLockType(redisLockType); + final Lock lock = registry.obtain("foo"); + + assertThat(lock.tryLock()).isTrue(); + try { + registry.renewLock("foo"); + } + finally { + lock.unlock(); + } + } + + @ParameterizedTest + @EnumSource(RedisLockType.class) + void testLockRenewLockNotOwned(RedisLockType redisLockType) { + final RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey); + registry.setRedisLockType(redisLockType); + registry.obtain("foo"); + + assertThatExceptionOfType(IllegalStateException.class) + .isThrownBy(() -> registry.renewLock("foo")); + } + @Test void testInitialiseWithCustomExecutor() { RedisLockRegistry redisLockRegistry = new RedisLockRegistry(redisConnectionFactory, "registryKey"); diff --git a/src/reference/antora/modules/ROOT/pages/redis.adoc b/src/reference/antora/modules/ROOT/pages/redis.adoc index 545777fb36..bead76431d 100644 --- a/src/reference/antora/modules/ROOT/pages/redis.adoc +++ b/src/reference/antora/modules/ROOT/pages/redis.adoc @@ -856,4 +856,7 @@ Default. The pub-sub is preferred mode - less network chatter between client Redis server, and more performant - the lock is acquired immediately when subscription is notified about unlocking in the other process. However, the Redis does not support pub-sub in the Master/Replica connections (for example in AWS ElastiCache environment), therefore a busy-spin mode is chosen as a default to make the registry working in any environment. -Starting with version 6.4, instead of throwing `IllegalStateException`, the `RedisLockRegistry.RedisLock.unlock()` method throws `ConcurrentModificationException` if the ownership of the lock is expired. \ No newline at end of file +Starting with version 6.4, instead of throwing `IllegalStateException`, the `RedisLockRegistry.RedisLock.unlock()` method throws `ConcurrentModificationException` if the ownership of the lock is expired. + +Starting with version 6.4, a `RedisLockRegistry.setRenewalTaskScheduler()` is added to configure the scheduler for periodic renewal of locks. +When it is set, the lock will be automatically renewed every `1/3` of the expiration time after the lock is successfully acquired, until unlocked or the redis key is removed. diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 3a0b552d4c..7b19af0bf0 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -59,6 +59,7 @@ See xref:zeromq.adoc[ZeroMQ Support] for more information. === Redis Changes Instead of throwing `IllegalStateException`, the `RedisLockRegistry.RedisLock.unlock()` method throws `ConcurrentModificationException` if the ownership of the lock is expired. +Add a `RedisLockRegistry.setRenewalTaskScheduler()` to periodic lock renewal. See xref:redis.adoc[Redis Support] for more information. [[x6.4-groovy-changes]]