Alawn/springside4
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

#525 加强Threads相关Utils类，增加ThreadDummper，加强ThreadPoolBuilder

Browse Source
This commit is contained in:
calvin.xiao
2016-12-29 18:39:07 +08:00
parent fd668c5098
commit 2beb80faaf
3 changed files with 177 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadDumper.java
+132
View File
@@ -0,0 +1,132 @@
package org.springside.modules.utils.thread;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 打印线程栈.
*
* 为避免打印过于频繁, 需要设置最小时间间隔, 默认一小时.
*
* 可设置StackTrace的深度，默认为8.
*/
public class ThreadDumper {
private static final int DEFAULT_LEAST_INTERVAL = 3600000; // 1小时
private static final int DEFAULT_MAX_STACK_LEVEL = 8;
private static final String DEFAULT_DUMP_REASON = "unknown reason";
private static Logger logger = LoggerFactory.getLogger(ThreadDumper.class);
private boolean enable = true; // 快速关闭该功能
private int leastInterval = DEFAULT_LEAST_INTERVAL; // 每次打印ThreadDump的最小时间间隔
private int maxStackLevel = DEFAULT_MAX_STACK_LEVEL; // 打印StackTrace的最大深度
private ThreadMXBean threadMBean = ManagementFactory.getThreadMXBean();
private AtomicLong lastThreadDumpTime = new AtomicLong(0);
/**
* 符合条件则打印线程栈.
*/
public void threadDumpIfNeed() {
threadDumpIfNeed(DEFAULT_DUMP_REASON);
}
/**
* 符合条件则打印线程栈.
*
* @param reasonMsg 发生ThreadDump的原因
*/
public void threadDumpIfNeed(String reasonMsg) {
if (!enable) {
return;
}
synchronized (lastThreadDumpTime) {
if (System.currentTimeMillis() - lastThreadDumpTime.get() < leastInterval) {
return;
} else {
lastThreadDumpTime.set(System.currentTimeMillis());
}
}
// 参数均为false, 避免输出lockedMonitors和lockedSynchronizers导致的JVM缓慢
ThreadInfo[] threadInfos = threadMBean.dumpAllThreads(false, false);
StringBuilder b = new StringBuilder(8192);
b.append('[');
for (int i = 0; i < threadInfos.length; i++) {
b.append(dumpThreadInfo(threadInfos[i])).append(", ");
}
logger.info("Dump thread info for {}", reasonMsg);
logger.info(b.toString());
}
/**
* 打印全部的stack，重新实现threadInfo的toString()函数，因为默认最多只打印8层的stack. 同时，不再打印lockedMonitors和lockedSynchronizers.
*/
public String dumpThreadInfo(ThreadInfo threadInfo) {
StringBuilder sb = new StringBuilder(512);
sb.append("\"").append(threadInfo.getThreadName()).append("\" Id=").append(threadInfo.getThreadId()).append(" ")
.append(threadInfo.getThreadState());
if (threadInfo.getLockName() != null) {
sb.append(" on ").append(threadInfo.getLockName());
}
if (threadInfo.getLockOwnerName() != null) {
sb.append(" owned by \"").append(threadInfo.getLockOwnerName()).append("\" Id=")
.append(threadInfo.getLockOwnerId());
}
if (threadInfo.isSuspended()) {
sb.append(" (suspended)");
}
if (threadInfo.isInNative()) {
sb.append(" (in native)");
}
sb.append('\n');
int i = 0;
StackTraceElement[] stackTrace = threadInfo.getStackTrace();
for (; i < Math.min(maxStackLevel, stackTrace.length); i++) {
StackTraceElement ste = stackTrace[i];
sb.append("\tat " + ste.toString()).append('\n');
if (i == 0 && threadInfo.getLockInfo() != null) {
Thread.State ts = threadInfo.getThreadState();
switch (ts) {
case BLOCKED:
sb.append("\t- blocked on ").append(threadInfo.getLockInfo()).append('\n');
break;
case WAITING:
sb.append("\t- waiting on ").append(threadInfo.getLockInfo()).append('\n');
break;
case TIMED_WAITING:
sb.append("\t- time waiting on ").append(threadInfo.getLockInfo()).append('\n');
break;
default:
}
}
}
if (i < stackTrace.length) {
sb.append("\t...").append('\n');
}
sb.append('\n');
return sb.toString();
}
public void setEnable(boolean enable) {
this.enable = enable;
}
public void setLeastInterval(int leastInterval) {
this.leastInterval = leastInterval;
}
public void setMaxStackLevel(int maxStackLevel) {
this.maxStackLevel = maxStackLevel;
}
}
modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadLocalContext.java
+2 -2
View File
@@ -16,14 +16,14 @@ public class ThreadLocalContext {
};
/**
* 放入ThreadLocal.
* 放入ThreadLocal的上下文信息.
*/
public static void put(String key, Object value) {
contextMap.get().put(key, value);
}
/**
* 取出ThreadLocal.
* 取出ThreadLocal的上下文信息.
*/
@SuppressWarnings("unchecked")
public static <T> T get(String key) {
modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadPoolBuilder.java
+43 -68
View File
@@ -49,6 +49,9 @@ public class ThreadPoolBuilder {
private int queueSize = 0;
private ThreadFactory threadFactory = null;
private String threadNamePrefix = null;
private Boolean daemon = null;
private RejectedExecutionHandler rejectHandler;
public FixedThreadPoolBuilder setPoolSize(int poolSize) {
@@ -61,11 +64,24 @@ public class ThreadPoolBuilder {
return this;
}
/**
* 与threadNamePrefix互斥, 优先使用ThreadFactory
*/
public FixedThreadPoolBuilder setThreadFactory(ThreadFactory threadFactory) {
this.threadFactory = threadFactory;
return this;
}
public FixedThreadPoolBuilder setThreadNamePrefix(String threadNamePrefix) {
this.threadNamePrefix = threadNamePrefix;
return this;
}
public FixedThreadPoolBuilder setdaemon(Boolean daemon) {
this.daemon = daemon;
return this;
}
public FixedThreadPoolBuilder setRejectHanlder(RejectedExecutionHandler rejectHandler) {
this.rejectHandler = rejectHandler;
return this;
@@ -84,7 +100,7 @@ public class ThreadPoolBuilder {
}
if (threadFactory == null) {
threadFactory = Executors.defaultThreadFactory();
threadFactory = createThreadFactory(threadNamePrefix, daemon);
}
if (rejectHandler == null) {
@@ -110,15 +126,19 @@ public class ThreadPoolBuilder {
* 2.b 如果设置了maxSize, 当总线程数达到上限, 会调用RejectHandler进行处理, 默认为AbortPolicy, 抛出RejectedExecutionException异常.
* 其他可选的Policy包括静默放弃当前任务(Discard)，或由主线程来直接执行(CallerRuns).
*
* 3. minSize以上, maxSize以下的线程, 如果在keepAliveTime中都poll不到任务执行将会被结束掉, keeAliveTime默认为60秒, 可设置.
* 3. minSize以上, maxSize以下的线程, 如果在keepAliveTime中都poll不到任务执行将会被结束掉, keeAliveTimeJDK默认为10秒.
* JDK默认值60秒太高，如高达1000线程，低于16QPS时才会回收开始回收.
*/
public static class CachedThreadPoolBuilder {
private int minSize = 0;
private int maxSize = Integer.MAX_VALUE;
private int keepAliveSecs = 60;
private int keepAliveSecs = 10;
private ThreadFactory threadFactory = null;
private String threadNamePrefix = null;
private Boolean daemon = null;
private RejectedExecutionHandler rejectHandler;
public CachedThreadPoolBuilder setMinSize(int minSize) {
@@ -136,11 +156,24 @@ public class ThreadPoolBuilder {
return this;
}
/**
* 与threadNamePrefix互斥, 优先使用ThreadFactory
*/
public CachedThreadPoolBuilder setThreadFactory(ThreadFactory threadFactory) {
this.threadFactory = threadFactory;
return this;
}
public CachedThreadPoolBuilder setThreadNamePrefix(String threadNamePrefix) {
this.threadNamePrefix = threadNamePrefix;
return this;
}
public CachedThreadPoolBuilder setdaemon(Boolean daemon) {
this.daemon = daemon;
return this;
}
public CachedThreadPoolBuilder setRejectHanlder(RejectedExecutionHandler rejectHandler) {
this.rejectHandler = rejectHandler;
return this;
@@ -149,7 +182,7 @@ public class ThreadPoolBuilder {
public ExecutorService build() {
if (threadFactory == null) {
threadFactory = Executors.defaultThreadFactory();
threadFactory = createThreadFactory(threadNamePrefix, daemon);
}
if (rejectHandler == null) {
@@ -161,73 +194,15 @@ public class ThreadPoolBuilder {
}
}
/**
* 可同时设置min/max/queue Size的线程池, 仅用于特殊场景.
*
* 比如并发要求非常高，觉得SynchronousQueue的性能太差.
*
* 比如平常使用Core线程工作，如果满了先放queue，queue满再开临时线程，此时queue的长度一定要按项目需求设好.
*/
public static class ConfigurableThreadPoolBuilder {
private int minSize = 0;
private int maxSize = Integer.MAX_VALUE;
private int queueSize = 0;
private int keepAliveSecs = 60;
private ThreadFactory threadFactory = null;
private RejectedExecutionHandler rejectHandler;
public ConfigurableThreadPoolBuilder setMinSize(int minSize) {
this.minSize = minSize;
return this;
}
public ConfigurableThreadPoolBuilder setMaxSize(int maxSize) {
this.maxSize = maxSize;
return this;
}
public ConfigurableThreadPoolBuilder setQueueSize(int queueSize) {
this.queueSize = queueSize;
return this;
}
public ConfigurableThreadPoolBuilder setKeepAliveSecs(int keepAliveSecs) {
this.keepAliveSecs = keepAliveSecs;
return this;
}
public ConfigurableThreadPoolBuilder setThreadFactory(ThreadFactory threadFactory) {
this.threadFactory = threadFactory;
return this;
}
public ConfigurableThreadPoolBuilder setRejectHanlder(RejectedExecutionHandler rejectHandler) {
this.rejectHandler = rejectHandler;
return this;
}
public ExecutorService build() {
BlockingQueue<Runnable> queue = null;
if (queueSize == 0) {
queue = new LinkedBlockingQueue<Runnable>();
private static ThreadFactory createThreadFactory(String threadNamePrefix, Boolean daemon) {
if (threadNamePrefix != null) {
if (daemon != null) {
return Threads.buildThreadFactory(threadNamePrefix + "-%d", daemon);
} else {
queue = new ArrayBlockingQueue<Runnable>(queueSize);
return Threads.buildThreadFactory(threadNamePrefix + "-%d");
}
if (threadFactory == null) {
threadFactory = Executors.defaultThreadFactory();
}
if (rejectHandler == null) {
rejectHandler = defaultRejectHandler;
}
return new ThreadPoolExecutor(minSize, maxSize, keepAliveSecs, TimeUnit.SECONDS, queue, threadFactory,
rejectHandler);
}
return Executors.defaultThreadFactory();
}
}