diff --git a/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadDumper.java b/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadDumper.java new file mode 100644 index 00000000..b5fa596f --- /dev/null +++ b/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadDumper.java @@ -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; + } +} diff --git a/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadLocalContext.java b/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadLocalContext.java index 082a5d82..96bb150a 100644 --- a/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadLocalContext.java +++ b/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadLocalContext.java @@ -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 get(String key) { diff --git a/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadPoolBuilder.java b/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadPoolBuilder.java index 55554066..d3a0f8c1 100644 --- a/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadPoolBuilder.java +++ b/modules/utils/src/main/java/org/springside/modules/utils/thread/ThreadPoolBuilder.java @@ -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 queue = null; - if (queueSize == 0) { - queue = new LinkedBlockingQueue(); + private static ThreadFactory createThreadFactory(String threadNamePrefix, Boolean daemon) { + if (threadNamePrefix != null) { + if (daemon != null) { + return Threads.buildThreadFactory(threadNamePrefix + "-%d", daemon); } else { - queue = new ArrayBlockingQueue(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(); } }