mirror of
https://github.com/spring-projects/spring-framework.git
synced 2026-10-03 13:39:34 +00:00
Add placeholder resolution support for @ConcurrencyLimit
See gh-35461 Closes gh-35470 Signed-off-by: Hyunsang Han <gustkd3@gmail.com>
This commit is contained in:
committed by
Sam Brannen
parent
23b8c61b99
commit
91f112a918
+9
@@ -42,6 +42,7 @@ import org.springframework.aot.hint.annotation.Reflective;
|
||||
* {@link org.springframework.aop.interceptor.ConcurrencyThrottleInterceptor}.
|
||||
*
|
||||
* @author Juergen Hoeller
|
||||
* @author Hyunsang Han
|
||||
* @since 7.0
|
||||
* @see EnableResilientMethods
|
||||
* @see ConcurrencyLimitBeanPostProcessor
|
||||
@@ -62,4 +63,12 @@ public @interface ConcurrencyLimit {
|
||||
*/
|
||||
int value() default 1;
|
||||
|
||||
/**
|
||||
* The concurrency limit as a configurable String.
|
||||
* A non-empty value specified here overrides the {@link #value()} attribute.
|
||||
* <p>This supports Spring-style "${...}" placeholders as well as SpEL expressions.
|
||||
* @see #value()
|
||||
*/
|
||||
String valueString() default "";
|
||||
|
||||
}
|
||||
|
||||
+29
-3
@@ -31,9 +31,12 @@ import org.springframework.aop.interceptor.ConcurrencyThrottleInterceptor;
|
||||
import org.springframework.aop.support.ComposablePointcut;
|
||||
import org.springframework.aop.support.DefaultPointcutAdvisor;
|
||||
import org.springframework.aop.support.annotation.AnnotationMatchingPointcut;
|
||||
import org.springframework.context.EmbeddedValueResolverAware;
|
||||
import org.springframework.core.annotation.AnnotatedElementUtils;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ConcurrentReferenceHashMap;
|
||||
import org.springframework.util.StringUtils;
|
||||
import org.springframework.util.StringValueResolver;
|
||||
|
||||
/**
|
||||
* A convenient {@link org.springframework.beans.factory.config.BeanPostProcessor
|
||||
@@ -41,10 +44,14 @@ import org.springframework.util.ConcurrentReferenceHashMap;
|
||||
* annotated with {@link ConcurrencyLimit @ConcurrencyLimit}.
|
||||
*
|
||||
* @author Juergen Hoeller
|
||||
* @author Hyunsang Han
|
||||
* @since 7.0
|
||||
*/
|
||||
@SuppressWarnings("serial")
|
||||
public class ConcurrencyLimitBeanPostProcessor extends AbstractBeanFactoryAwareAdvisingPostProcessor {
|
||||
public class ConcurrencyLimitBeanPostProcessor extends AbstractBeanFactoryAwareAdvisingPostProcessor
|
||||
implements EmbeddedValueResolverAware {
|
||||
|
||||
private @Nullable StringValueResolver embeddedValueResolver;
|
||||
|
||||
public ConcurrencyLimitBeanPostProcessor() {
|
||||
setBeforeExistingAdvisors(true);
|
||||
@@ -57,7 +64,13 @@ public class ConcurrencyLimitBeanPostProcessor extends AbstractBeanFactoryAwareA
|
||||
}
|
||||
|
||||
|
||||
private static class ConcurrencyLimitInterceptor implements MethodInterceptor {
|
||||
@Override
|
||||
public void setEmbeddedValueResolver(StringValueResolver resolver) {
|
||||
this.embeddedValueResolver = resolver;
|
||||
}
|
||||
|
||||
|
||||
private class ConcurrencyLimitInterceptor implements MethodInterceptor {
|
||||
|
||||
private final Map<Object, ConcurrencyThrottleCache> cachePerInstance =
|
||||
new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);
|
||||
@@ -93,7 +106,8 @@ public class ConcurrencyLimitBeanPostProcessor extends AbstractBeanFactoryAwareA
|
||||
}
|
||||
if (interceptor == null) {
|
||||
Assert.state(limit != null, "No @ConcurrencyLimit annotation found");
|
||||
interceptor = new ConcurrencyThrottleInterceptor(limit.value());
|
||||
int concurrencyLimit = parseInt(limit.value(), limit.valueString());
|
||||
interceptor = new ConcurrencyThrottleInterceptor(concurrencyLimit);
|
||||
if (!perMethod) {
|
||||
cache.classInterceptor = interceptor;
|
||||
}
|
||||
@@ -104,6 +118,18 @@ public class ConcurrencyLimitBeanPostProcessor extends AbstractBeanFactoryAwareA
|
||||
}
|
||||
return interceptor.invoke(invocation);
|
||||
}
|
||||
|
||||
private int parseInt(int value, String stringValue) {
|
||||
if (StringUtils.hasText(stringValue)) {
|
||||
if (embeddedValueResolver != null) {
|
||||
stringValue = embeddedValueResolver.resolveStringValue(stringValue);
|
||||
}
|
||||
if (StringUtils.hasText(stringValue)) {
|
||||
return Integer.parseInt(stringValue);
|
||||
}
|
||||
}
|
||||
return value;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
+52
@@ -18,6 +18,7 @@ package org.springframework.resilience;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Properties;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
@@ -28,13 +29,17 @@ import org.springframework.aop.framework.ProxyFactory;
|
||||
import org.springframework.aop.interceptor.ConcurrencyThrottleInterceptor;
|
||||
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
|
||||
import org.springframework.beans.factory.support.RootBeanDefinition;
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import org.springframework.core.env.PropertiesPropertySource;
|
||||
import org.springframework.resilience.annotation.ConcurrencyLimit;
|
||||
import org.springframework.resilience.annotation.ConcurrencyLimitBeanPostProcessor;
|
||||
import org.springframework.resilience.annotation.EnableResilientMethods;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Juergen Hoeller
|
||||
* @author Hyunsang Han
|
||||
* @since 7.0
|
||||
*/
|
||||
class ConcurrencyLimitTests {
|
||||
@@ -97,6 +102,28 @@ class ConcurrencyLimitTests {
|
||||
assertThat(target.current).hasValue(0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withPlaceholderResolution() {
|
||||
Properties props = new Properties();
|
||||
props.setProperty("test.concurrency.limit", "3");
|
||||
|
||||
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
|
||||
ctx.getEnvironment().getPropertySources().addFirst(new PropertiesPropertySource("test", props));
|
||||
ctx.register(PlaceholderTestConfig.class, PlaceholderBean.class);
|
||||
ctx.refresh();
|
||||
|
||||
PlaceholderBean proxy = ctx.getBean(PlaceholderBean.class);
|
||||
PlaceholderBean target = (PlaceholderBean) AopProxyUtils.getSingletonTarget(proxy);
|
||||
|
||||
// Test with limit=3 from properties
|
||||
List<CompletableFuture<?>> futures = new ArrayList<>(10);
|
||||
for (int i = 0; i < 10; i++) {
|
||||
futures.add(CompletableFuture.runAsync(proxy::concurrentOperation));
|
||||
}
|
||||
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
|
||||
assertThat(target.current).hasValue(0);
|
||||
ctx.close();
|
||||
}
|
||||
|
||||
static class NonAnnotatedBean {
|
||||
|
||||
@@ -185,4 +212,29 @@ class ConcurrencyLimitTests {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@EnableResilientMethods
|
||||
static class PlaceholderTestConfig {
|
||||
}
|
||||
|
||||
|
||||
static class PlaceholderBean {
|
||||
|
||||
AtomicInteger current = new AtomicInteger();
|
||||
|
||||
@ConcurrencyLimit(valueString = "${test.concurrency.limit}")
|
||||
public void concurrentOperation() {
|
||||
if (current.incrementAndGet() > 3) { // Assumes test.concurrency.limit=3
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
try {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
throw new IllegalStateException(ex);
|
||||
}
|
||||
current.decrementAndGet();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user