anph
/
hugegraph-sync
forked from hugegraph/hugegraph-sync
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(commons):fixed memory leaks occur in HugeGraph Server during data writing (#144) 

* #2578
fixed memory leaks occur in HugeGraph Server during data writing
This commit is contained in:
haohao0103 2024-07-30 17:46:59 +08:00 committed by GitHub
parent 93c2e08780
commit 9baef8b255
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 64 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java
View File

@ -57,16 +57,23 @@ public class EventHub {
}
public EventHub(String name) {
this(name, 1);
this(name, 1, Runtime.getRuntime().availableProcessors() << 2);
}
public EventHub(String name, int threadSize) {
LOG.debug("Create new EventHub {}", name);
LOG.debug("Create new EventHub {},threadSize {}", name, threadSize);
this.name = name;
this.listeners = new ConcurrentHashMap<>();
EventHub.init(threadSize);
}
public EventHub(String name, int corePoolSize, int maximumPoolSize) {
LOG.debug("Create new EventHub {},corePoolSize {}, maximumPoolSize {}", name, corePoolSize, maximumPoolSize);
this.name = name;
this.listeners = new ConcurrentHashMap<>();
EventHub.init(corePoolSize, maximumPoolSize);
}
public static synchronized void init(int poolSize) {
if (executor != null) {
return;
@ -75,6 +82,15 @@ public class EventHub {
executor = ExecutorUtil.newFixedThreadPool(poolSize, EVENT_WORKER);
}
public static synchronized void init(int corePoolSize, int maximumPoolSize) {
LOG.debug("Init corePoolSize {}, maximumPoolSize {} for EventHub", corePoolSize, maximumPoolSize);
if (executor != null) {
LOG.debug("EventHub executor already initialized");
return;
}
executor = ExecutorUtil.newDynamicThreadExecutor(EVENT_WORKER, corePoolSize, maximumPoolSize);
}
public static synchronized boolean destroy(long timeout)
throws InterruptedException {
E.checkState(executor != null, "EventHub has not been initialized");

47
hugegraph-common/src/main/java/org/apache/hugegraph/util/ExecutorUtil.java
View File

@ -21,13 +21,58 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.hugegraph.concurrent.PausableScheduledThreadPool;
public final class ExecutorUtil {
public static ThreadPoolExecutor newDynamicThreadExecutor(String name,
int corePoolSize,
int maximumPoolSize) {
long keepAliveTime = 60L;
TimeUnit unit = TimeUnit.SECONDS;
ThreadFactory factory = new BasicThreadFactory.Builder()
.namingPattern(name)
.build();
CustomBlockingQueue<Runnable> workQueue = new CustomBlockingQueue<>();
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime,
unit,
workQueue,
factory,
new ThreadPoolExecutor.CallerRunsPolicy()
);
workQueue.setThreadPoolExecutor(threadPoolExecutor);
return threadPoolExecutor;
}
static class CustomBlockingQueue<E> extends LinkedBlockingQueue<E> {
private ThreadPoolExecutor threadPoolExecutor;
public void setThreadPoolExecutor(ThreadPoolExecutor threadPoolExecutor) {
this.threadPoolExecutor = threadPoolExecutor;
}
@Override
public boolean offer(E e) {
if (threadPoolExecutor.getPoolSize() < threadPoolExecutor.getMaximumPoolSize()) {
return false;
}
return super.offer(e);
}
}
public static ExecutorService newFixedThreadPool(String name) {
return newFixedThreadPool(1, name);
}