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

GH-3672: Clean up Jdbc & ZK LockRegistry caches

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3672

* Clean up `JdbcLockRegistry`, `ZookeeperLockRegistry` cache automatically 
* setCapacity(int capacity) to cacheCapacity(int capacity)
* field rename `capacity`to `cacheCapacity`, add static
This commit is contained in:
Unseok Kim
2021-11-12 06:36:27 +09:00
committed by GitHub
parent 5452a6fbe6
commit db611028da
9 changed files with 447 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java
View File

@@ -17,10 +17,9 @@
package org.springframework.integration.jdbc.lock;
import java.time.Duration;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
@@ -54,6 +53,7 @@ import org.springframework.util.Assert;
* @author Stefan Vassilev
* @author Olivier Hubaut
* @author Fran Aranda
* @author Unseok Kim
*
* @since 4.3
*/
@@ -61,12 +61,24 @@ public class JdbcLockRegistry implements ExpirableLockRegistry, RenewableLockReg
private static final int DEFAULT_IDLE = 100;
private final Map<String, JdbcLock> locks = new ConcurrentHashMap<>();
private static final int DEFAULT_CAPACITY = 100_000;
private final Map<String, JdbcLock> locks =
new LinkedHashMap<String, JdbcLock>(16, 0.75F, true) {
@Override
protected boolean removeEldestEntry(Entry<String, JdbcLock> eldest) {
return size() > JdbcLockRegistry.this.cacheCapacity;
}
};
private final LockRepository client;
private Duration idleBetweenTries = Duration.ofMillis(DEFAULT_IDLE);
private int cacheCapacity = DEFAULT_CAPACITY;
public JdbcLockRegistry(LockRepository client) {
this.client = client;
}
@@ -82,11 +94,22 @@ public class JdbcLockRegistry implements ExpirableLockRegistry, RenewableLockReg
this.idleBetweenTries = idleBetweenTries;
}
/**
* Set the capacity of cached locks.
* @param cacheCapacity The capacity of cached lock, (default 100_000).
* @since 5.5.6
*/
public void setCacheCapacity(int cacheCapacity) {
this.cacheCapacity = cacheCapacity;
}
@Override
public Lock obtain(Object lockKey) {
Assert.isInstanceOf(String.class, lockKey);
String path = pathFor((String) lockKey);
return this.locks.computeIfAbsent(path, (key) -> new JdbcLock(this.client, this.idleBetweenTries, key));
synchronized (this.locks) {
return this.locks.computeIfAbsent(path, key -> new JdbcLock(this.client, this.idleBetweenTries, key));
}
}
private String pathFor(String input) {
@@ -95,14 +118,13 @@ public class JdbcLockRegistry implements ExpirableLockRegistry, RenewableLockReg
@Override
public void expireUnusedOlderThan(long age) {
Iterator<Entry<String, JdbcLock>> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
while (iterator.hasNext()) {
Entry<String, JdbcLock> entry = iterator.next();
JdbcLock lock = entry.getValue();
if (now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess()) {
iterator.remove();
}
synchronized (this.locks) {
this.locks.entrySet()
.removeIf(entry -> {
JdbcLock lock = entry.getValue();
return now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess();
});
}
}
@@ -110,7 +132,10 @@ public class JdbcLockRegistry implements ExpirableLockRegistry, RenewableLockReg
public void renewLock(Object lockKey) {
Assert.isInstanceOf(String.class, lockKey);
String path = pathFor((String) lockKey);
JdbcLock jdbcLock = this.locks.get(path);
JdbcLock jdbcLock;
synchronized (this.locks) {
jdbcLock = this.locks.get(path);
}
if (jdbcLock == null) {
throw new IllegalStateException("Could not found mutex at " + path);
}

175
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2021 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.
@@ -20,8 +20,12 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
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.locks.Lock;
@@ -35,6 +39,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@@ -43,6 +48,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
* @author Artem Bilan
* @author Stefan Vassilev
* @author Alexandre Strubel
* @author Unseok Kim
*
* @since 4.3
*/
@@ -312,4 +318,171 @@ public class JdbcLockRegistryTests {
.isThrownBy(() -> registry.renewLock("foo"));
}
@Test
public void concurrentObtainCapacityTest() throws InterruptedException {
final int KEY_CNT = 500;
final int CAPACITY_CNT = 179;
final int THREAD_CNT = 4;
final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT);
registry.setCacheCapacity(CAPACITY_CNT);
final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT);
for (int i = 0; i < KEY_CNT; i++) {
int finalI = i;
executorService.submit(() -> {
countDownLatch.countDown();
try {
countDownLatch.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
String keyId = "foo:" + finalI;
Lock obtain = registry.obtain(keyId);
obtain.lock();
obtain.unlock();
});
}
executorService.shutdown();
executorService.awaitTermination(5, TimeUnit.SECONDS);
//capacity limit test
assertThat(getRegistryLocks(registry)).hasSize(CAPACITY_CNT);
registry.expireUnusedOlderThan(-1000);
assertThat(getRegistryLocks(registry)).isEmpty();
}
@Test
public void concurrentObtainRemoveOrderTest() throws InterruptedException {
final int THREAD_CNT = 2;
final int DUMMY_LOCK_CNT = 3;
final int CAPACITY_CNT = THREAD_CNT;
final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT);
registry.setCacheCapacity(CAPACITY_CNT);
final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT);
final Queue<String> remainLockCheckQueue = new LinkedBlockingQueue<>();
//Removed due to capcity limit
for (int i = 0; i < DUMMY_LOCK_CNT; i++) {
Lock obtainLock0 = registry.obtain("foo:" + i);
obtainLock0.lock();
obtainLock0.unlock();
}
for (int i = DUMMY_LOCK_CNT; i < THREAD_CNT + DUMMY_LOCK_CNT; i++) {
int finalI = i;
executorService.submit(() -> {
countDownLatch.countDown();
try {
countDownLatch.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
String keyId = "foo:" + finalI;
remainLockCheckQueue.offer(toUUID(keyId));
Lock obtain = registry.obtain(keyId);
obtain.lock();
obtain.unlock();
});
}
executorService.shutdown();
executorService.awaitTermination(5, TimeUnit.SECONDS);
assertThat(getRegistryLocks(registry)).containsKeys(
remainLockCheckQueue.toArray(new String[remainLockCheckQueue.size()]));
}
@Test
public void concurrentObtainAccessRemoveOrderTest() throws InterruptedException {
final int THREAD_CNT = 2;
final int DUMMY_LOCK_CNT = 3;
final int CAPACITY_CNT = THREAD_CNT + 1;
final String REMAIN_DUMMY_LOCK_KEY = "foo:1";
final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT);
registry.setCacheCapacity(CAPACITY_CNT);
final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT);
final Queue<String> remainLockCheckQueue = new LinkedBlockingQueue<>();
//Removed due to capcity limit
for (int i = 0; i < DUMMY_LOCK_CNT; i++) {
Lock obtainLock0 = registry.obtain("foo:" + i);
obtainLock0.lock();
obtainLock0.unlock();
}
Lock obtainLock0 = registry.obtain(REMAIN_DUMMY_LOCK_KEY);
obtainLock0.lock();
obtainLock0.unlock();
remainLockCheckQueue.offer(toUUID(REMAIN_DUMMY_LOCK_KEY));
for (int i = DUMMY_LOCK_CNT; i < THREAD_CNT + DUMMY_LOCK_CNT; i++) {
int finalI = i;
executorService.submit(() -> {
countDownLatch.countDown();
try {
countDownLatch.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
String keyId = "foo:" + finalI;
remainLockCheckQueue.offer(toUUID(keyId));
Lock obtain = registry.obtain(keyId);
obtain.lock();
obtain.unlock();
});
}
executorService.shutdown();
executorService.awaitTermination(5, TimeUnit.SECONDS);
assertThat(getRegistryLocks(registry)).containsKeys(
remainLockCheckQueue.toArray(new String[remainLockCheckQueue.size()]));
}
@Test
public void setCapacityTest() {
final int CAPACITY_CNT = 4;
registry.setCacheCapacity(CAPACITY_CNT);
registry.obtain("foo:1");
registry.obtain("foo:2");
registry.obtain("foo:3");
//capacity 4->3
registry.setCacheCapacity(CAPACITY_CNT - 1);
registry.obtain("foo:4");
assertThat(getRegistryLocks(registry)).hasSize(3);
assertThat(getRegistryLocks(registry)).containsKeys(toUUID("foo:2"),
toUUID("foo:3"),
toUUID("foo:4"));
//capacity 3->4
registry.setCacheCapacity(CAPACITY_CNT);
registry.obtain("foo:5");
assertThat(getRegistryLocks(registry)).hasSize(4);
assertThat(getRegistryLocks(registry)).containsKeys(toUUID("foo:3"),
toUUID("foo:4"),
toUUID("foo:5"));
}
@SuppressWarnings("unchecked")
private static Map<String, Lock> getRegistryLocks(JdbcLockRegistry registry) {
return TestUtils.getPropertyValue(registry, "locks", Map.class);
}
private static String toUUID(String key) {
return UUIDConverter.getUUID(key).toString();
}
}