This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 71baadfcca perf(cache): keep memory cache lookups on the subscribing
thread (#7849)
71baadfcca is described below
commit 71baadfcca6ec2b71b41eb45e2d5ea4d8476e502
Author: hanzhe-one <[email protected]>
AuthorDate: Mon Oct 19 14:00:46 2026 +0800
perf(cache): keep memory cache lookups on the subscribing thread (#7849)
MemoryCache is a local Caffeine cache, but every read wrapped the lookup
in Mono.fromCallable(...).subscribeOn(Schedulers.boundedElastic()), so a
cache hit queued a task on the bounded-elastic pool even though the work
was a local map read. cacheData also scheduled an already-computed
Boolean.TRUE the same way.
Drop the unconditional scheduler hand-off from cacheData, isExist and
getData. The lookups stay lazy at subscription time via fromCallable, so
expiry and miss semantics are unchanged; only the thread the lookup runs
on differs. Callers that need a different thread can still apply their
own subscribeOn.
---
.../shenyu/plugin/cache/memory/MemoryCache.java | 9 +++--
.../plugin/cache/memory/MemoryCacheTest.java | 43 ++++++++++++++++++++++
2 files changed, 48 insertions(+), 4 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-memory/src/main/java/org/apache/shenyu/plugin/cache/memory/MemoryCache.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-memory/src/main/java/org/apache/shenyu/plugin/cache/memory/MemoryCache.java
index 97e7073d86..3afff45b7b 100644
---
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-memory/src/main/java/org/apache/shenyu/plugin/cache/memory/MemoryCache.java
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-memory/src/main/java/org/apache/shenyu/plugin/cache/memory/MemoryCache.java
@@ -23,7 +23,6 @@ import com.github.benmanes.caffeine.cache.Expiry;
import com.github.benmanes.caffeine.cache.Scheduler;
import org.apache.shenyu.plugin.cache.ICache;
import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
@@ -71,7 +70,7 @@ public final class MemoryCache implements ICache {
this.mainCache.policy().expireVariably()
.orElseThrow(() -> new IllegalStateException("Variable
expiration is not configured for the memory cache"))
.put(key, bytes, timeoutSeconds, TimeUnit.SECONDS);
- return Mono.fromCallable(() ->
Boolean.TRUE).subscribeOn(Schedulers.boundedElastic());
+ return Mono.just(Boolean.TRUE);
}
/**
@@ -82,7 +81,9 @@ public final class MemoryCache implements ICache {
*/
@Override
public Mono<Boolean> isExist(final String key) {
- return Mono.fromCallable(() ->
Objects.nonNull(this.mainCache.getIfPresent(key))).subscribeOn(Schedulers.boundedElastic());
+ // Deferred until subscription, but evaluated on the subscribing
thread: a local map read
+ // does not warrant a hand-off to another scheduler.
+ return Mono.fromCallable(() ->
Objects.nonNull(this.mainCache.getIfPresent(key)));
}
/**
@@ -93,7 +94,7 @@ public final class MemoryCache implements ICache {
*/
@Override
public Mono<byte[]> getData(final String key) {
- return Mono.fromCallable(() ->
this.mainCache.getIfPresent(key)).subscribeOn(Schedulers.boundedElastic());
+ return Mono.fromCallable(() -> this.mainCache.getIfPresent(key));
}
/**
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-memory/src/test/java/org/apache/shenyu/plugin/cache/memory/MemoryCacheTest.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-memory/src/test/java/org/apache/shenyu/plugin/cache/memory/MemoryCacheTest.java
index 132f485c4a..36b03d3d74 100644
---
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-memory/src/test/java/org/apache/shenyu/plugin/cache/memory/MemoryCacheTest.java
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-memory/src/test/java/org/apache/shenyu/plugin/cache/memory/MemoryCacheTest.java
@@ -21,13 +21,18 @@ import com.github.benmanes.caffeine.cache.Cache;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.test.util.ReflectionTestUtils;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.awaitility.Awaitility.await;
/**
@@ -35,6 +40,8 @@ import static org.awaitility.Awaitility.await;
*/
public class MemoryCacheTest {
+ private static final String CALLER_THREAD_NAME = "caller-thread";
+
private final MemoryCache memoryCache = new MemoryCache();
@AfterEach
@@ -65,6 +72,30 @@ public class MemoryCacheTest {
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(0L, cache.estimatedSize()));
}
+ @Test
+ public void testReadsRunOnTheSubscribingThread() {
+ final String key = "data";
+ final byte[] bytes = "data".getBytes(StandardCharsets.UTF_8);
+ final Scheduler caller = Schedulers.newSingle(CALLER_THREAD_NAME);
+ try {
+ memoryCache.cacheData(key, bytes, 10).block();
+ assertSubscribesOnCallerThread(caller, memoryCache.isExist(key),
Boolean.TRUE);
+ assertSubscribesOnCallerThread(caller, memoryCache.getData(key),
bytes);
+ } finally {
+ caller.dispose();
+ }
+ }
+
+ @Test
+ public void testReadsAreDeferredUntilSubscription() {
+ final String key = "data";
+ final byte[] bytes = "data".getBytes(StandardCharsets.UTF_8);
+ final Mono<byte[]> deferred = memoryCache.getData(key);
+ // Assembling the publisher must not observe the cache; only
subscribing may.
+ memoryCache.cacheData(key, bytes, 10).block();
+ assertArrayEquals(bytes, deferred.block());
+ }
+
@Test
public void testPerEntryExpirationAndReplacement() {
final byte[] bytes = "data".getBytes(StandardCharsets.UTF_8);
@@ -84,4 +115,16 @@ public class MemoryCacheTest {
assertArrayEquals(bytes, memoryCache.getData("extended").block());
}
+ private <T> void assertSubscribesOnCallerThread(final Scheduler caller,
final Mono<T> publisher, final T expected) {
+ final AtomicReference<Thread> lookupThread = new AtomicReference<>();
+ final T actual = publisher
+ // Captures the thread that evaluates the deferred lookup,
which is the subscribing thread.
+ .doOnNext(ignored -> lookupThread.set(Thread.currentThread()))
+ .subscribeOn(caller)
+ .block();
+ assertEquals(expected, actual);
+ // The lookup must stay on the caller-supplied scheduler instead of
hopping to bounded-elastic.
+
assertTrue(lookupThread.get().getName().startsWith(CALLER_THREAD_NAME),
lookupThread.get().getName());
+ }
+
}
