lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

CacheMemorySource

Browse Source
This commit is contained in:
redkale
2023-09-07 16:40:52 +08:00
parent da41c7f7fa
commit 388505c899
1 changed files with 13 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/main/java/org/redkale/source/CacheMemorySource.java
View File

@@ -67,9 +67,9 @@ public final class CacheMemorySource extends AbstractCacheSource {
//key: topic //key: topic
private final Map<String, Set<CacheEventListener<byte[]>>> pubsubListeners = new ConcurrentHashMap<>(); private final Map<String, Set<CacheEventListener<byte[]>>> pubsubListeners = new ConcurrentHashMap<>();
private ExecutorService pubsubExecutor; private ExecutorService subExecutor;
private final ReentrantLock pubsubExecutorLock = new ReentrantLock(); private final ReentrantLock subExecutorLock = new ReentrantLock();
public CacheMemorySource(String resourceName) { public CacheMemorySource(String resourceName) {
this.name = resourceName; this.name = resourceName;
@@ -85,18 +85,18 @@ public final class CacheMemorySource extends AbstractCacheSource {
public void onResourceChange(ResourceEvent[] events) { public void onResourceChange(ResourceEvent[] events) {
} }
private ExecutorService pubsubExecutor() { protected ExecutorService subExecutor() {
ExecutorService executor = pubsubExecutor; ExecutorService executor = subExecutor;
if (executor != null) { if (executor != null) {
return executor; return executor;
} }
pubsubExecutorLock.lock(); subExecutorLock.lock();
try { try {
if (pubsubExecutor == null) { if (subExecutor == null) {
String threadNameFormat = "CacheSource-" + resourceName() + "-PubSubThread-%s"; String threadNameFormat = "CacheSource-" + resourceName() + "-SubThread-%s";
Function<String, ExecutorService> func = Utility.virtualExecutorFunction(); Function<String, ExecutorService> func = Utility.virtualExecutorFunction();
final AtomicInteger counter = new AtomicInteger(); final AtomicInteger counter = new AtomicInteger();
pubsubExecutor = func == null ? Executors.newFixedThreadPool(Utility.cpus(), r -> { subExecutor = func == null ? Executors.newFixedThreadPool(Utility.cpus(), r -> {
Thread t = new Thread(r); Thread t = new Thread(r);
t.setDaemon(true); t.setDaemon(true);
int c = counter.incrementAndGet(); int c = counter.incrementAndGet();
@@ -104,9 +104,9 @@ public final class CacheMemorySource extends AbstractCacheSource {
return t; return t;
}) : func.apply(threadNameFormat); }) : func.apply(threadNameFormat);
} }
executor = pubsubExecutor; executor = subExecutor;
} finally { } finally {
pubsubExecutorLock.unlock(); subExecutorLock.unlock();
} }
return executor; return executor;
} }
@@ -187,8 +187,8 @@ public final class CacheMemorySource extends AbstractCacheSource {
if (scheduler != null) { if (scheduler != null) {
scheduler.shutdownNow(); scheduler.shutdownNow();
} }
if (pubsubExecutor != null) { if (subExecutor != null) {
pubsubExecutor.shutdownNow(); subExecutor.shutdownNow();
} }
} }
@@ -254,7 +254,7 @@ public final class CacheMemorySource extends AbstractCacheSource {
if (listeners == null || listeners.isEmpty()) { if (listeners == null || listeners.isEmpty()) {
return CompletableFuture.completedFuture(0); return CompletableFuture.completedFuture(0);
} }
Executor executor = pubsubExecutor(); Executor executor = subExecutor();
listeners.forEach(listener -> executor.execute(() -> { listeners.forEach(listener -> executor.execute(() -> {
try { try {
listener.onMessage(topic, message); listener.onMessage(topic, message);