Dubbo里面线程池的拒绝策略
代码语言：javascript复制public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
protected static final Logger logger = LoggerFactory.getLogger(AbortPolicyWithReport.class);
private final String threadName;
private final URL url;
private static volatile long lastPrintTime = 0;
private static Semaphore guard = new Semaphore(1);
public AbortPolicyWithReport(String threadName, URL url) {
this.threadName = threadName;
this.url = url;
}
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
String msg = String.format("Thread pool is EXHAUSTED!"
" Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d),"
" Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
url.getProtocol(), url.getIp(), url.getPort());
logger.warn(msg);
dumpJStack();
throw new RejectedExecutionException(msg);
}
private void dumpJStack() {
long now = System.currentTimeMillis();
//dump every 10 minutes
//每 10 分钟，只打印一次。
if (now - lastPrintTime < 10 * 60 * 1000) {
return;
}
//获得信号量。保证，同一时间，有且仅有一个线程执行打印。
if (!guard.tryAcquire()) {
return;
}
//// 创建线程池，后台执行打印 JStack
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
// 获得路径
String dumpPath = url.getParameter(Constants.DUMP_DIRECTORY, System.getProperty("user.home"));
SimpleDateFormat sdf;
// 获得系统
String OS = System.getProperty("os.name").toLowerCase();
// window system don't support ":" in file name
if(OS.contains("win")){
sdf = new SimpleDateFormat("yyyy-MM-dd_HH-mm-ss");
}else {
sdf = new SimpleDateFormat("yyyy-MM-dd_HH:mm:ss");
}
String dateStr = sdf.format(new Date());
// 获得输出流
FileOutputStream jstackStream = null;
try {
jstackStream = new FileOutputStream(new File(dumpPath, "Dubbo_JStack.log" "." dateStr));
// 打印 JStack
JVMUtil.jstack(jstackStream);
} catch (Throwable t) {
logger.error("dump jstack error", t);
} finally {
// 释放信号量
guard.release();
// 释放输出流
if (jstackStream != null) {
try {
jstackStream.flush();
jstackStream.close();
} catch (IOException e) {
}
}
}
// 记录最后打印时间
lastPrintTime = System.currentTimeMillis();
}
});
}
}public class JVMUtil {
public static void jstack(OutputStream stream) throws Exception {
ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean();
for (ThreadInfo threadInfo : threadMxBean.dumpAllThreads(true, true)) {
stream.write(getThreadDumpString(threadInfo).getBytes());
}
}
private static String getThreadDumpString(ThreadInfo threadInfo) {
StringBuilder sb = new StringBuilder(""" threadInfo.getThreadName() """
" Id=" threadInfo.getThreadId() " "
threadInfo.getThreadState());
if (threadInfo.getLockName() != null) {
sb.append(" on " threadInfo.getLockName());
}
if (threadInfo.getLockOwnerName() != null) {
sb.append(" owned by "" threadInfo.getLockOwnerName()
"" Id=" 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();
MonitorInfo[] lockedMonitors = threadInfo.getLockedMonitors();
for (; i < stackTrace.length && i < 32; i ) {
StackTraceElement ste = stackTrace[i];
sb.append("tat " ste.toString());
sb.append('n');
if (i == 0 && threadInfo.getLockInfo() != null) {
Thread.State ts = threadInfo.getThreadState();
switch (ts) {
case BLOCKED:
sb.append("t- blocked on " threadInfo.getLockInfo());
sb.append('n');
break;
case WAITING:
sb.append("t- waiting on " threadInfo.getLockInfo());
sb.append('n');
break;
case TIMED_WAITING:
sb.append("t- waiting on " threadInfo.getLockInfo());
sb.append('n');
break;
default:
}
}
for (MonitorInfo mi : lockedMonitors) {
if (mi.getLockedStackDepth() == i) {
sb.append("t- locked " mi);
sb.append('n');
}
}
}
if (i < stackTrace.length) {
sb.append("t...");
sb.append('n');
}
LockInfo[] locks = threadInfo.getLockedSynchronizers();
if (locks.length > 0) {
sb.append("ntNumber of locked synchronizers = " locks.length);
sb.append('n');
for (LockInfo li : locks) {
sb.append("t- " li);
sb.append('n');
}
}
sb.append('n');
return sb.toString();
}
}
