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

Use ReentrantLock instead of synchronization for concurrency throttle

Browse Source 
Closes gh-32251
This commit is contained in:
Juergen Hoeller
2024-02-13 11:03:40 +01:00
parent 120ea0a51c
commit 0b09f1e12f
1 changed files with 19 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
spring-core/src/main/java/org/springframework/util/ConcurrencyThrottleSupport.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2023 the original author or authors. * Copyright 2002-2024 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -19,6 +19,9 @@ package org.springframework.util;
import java.io.IOException; import java.io.IOException;
import java.io.ObjectInputStream; import java.io.ObjectInputStream;
import java.io.Serializable; import java.io.Serializable;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
@@ -60,7 +63,9 @@ public abstract class ConcurrencyThrottleSupport implements Serializable {
/** Transient to optimize serialization. */ /** Transient to optimize serialization. */
protected transient Log logger = LogFactory.getLog(getClass()); protected transient Log logger = LogFactory.getLog(getClass());
private transient Object monitor = new Object(); private final Lock concurrencyLock = new ReentrantLock();
private final Condition concurrencyCondition = this.concurrencyLock.newCondition();
private int concurrencyLimit = UNBOUNDED_CONCURRENCY; private int concurrencyLimit = UNBOUNDED_CONCURRENCY;
@@ -109,7 +114,8 @@ public abstract class ConcurrencyThrottleSupport implements Serializable {
} }
if (this.concurrencyLimit > 0) { if (this.concurrencyLimit > 0) {
boolean debug = logger.isDebugEnabled(); boolean debug = logger.isDebugEnabled();
synchronized (this.monitor) { this.concurrencyLock.lock();
try {
boolean interrupted = false; boolean interrupted = false;
while (this.concurrencyCount >= this.concurrencyLimit) { while (this.concurrencyCount >= this.concurrencyLimit) {
if (interrupted) { if (interrupted) {
@@ -121,7 +127,7 @@ public abstract class ConcurrencyThrottleSupport implements Serializable {
" has reached limit " + this.concurrencyLimit + " - blocking"); " has reached limit " + this.concurrencyLimit + " - blocking");
} }
try { try {
this.monitor.wait(); this.concurrencyCondition.await();
} }
catch (InterruptedException ex) { catch (InterruptedException ex) {
// Re-interrupt current thread, to allow other threads to react. // Re-interrupt current thread, to allow other threads to react.
@@ -134,6 +140,9 @@ public abstract class ConcurrencyThrottleSupport implements Serializable {
} }
this.concurrencyCount++; this.concurrencyCount++;
} }
finally {
this.concurrencyLock.unlock();
}
} }
} }
@@ -144,12 +153,16 @@ public abstract class ConcurrencyThrottleSupport implements Serializable {
protected void afterAccess() { protected void afterAccess() {
if (this.concurrencyLimit >= 0) { if (this.concurrencyLimit >= 0) {
boolean debug = logger.isDebugEnabled(); boolean debug = logger.isDebugEnabled();
synchronized (this.monitor) { this.concurrencyLock.lock();
try {
this.concurrencyCount--; this.concurrencyCount--;
if (debug) { if (debug) {
logger.debug("Returning from throttle at concurrency count " + this.concurrencyCount); logger.debug("Returning from throttle at concurrency count " + this.concurrencyCount);
} }
this.monitor.notify(); this.concurrencyCondition.signal();
}
finally {
this.concurrencyLock.unlock();
} }
} }
} }
@@ -165,7 +178,6 @@ public abstract class ConcurrencyThrottleSupport implements Serializable {
// Initialize transient fields. // Initialize transient fields.
this.logger = LogFactory.getLog(getClass()); this.logger = LogFactory.getLog(getClass());
this.monitor = new Object();
} }
} }