dubbo生产者预热不充分.导致线程池耗尽
### 原因分析
这种情况就是提供者重启后预热不充分即被调用。因为当生产者刚启动时需要预热，需要和其它资源例如数据库、缓存等建立连接，建立连接是需要时间的。如果此时大量消费者请求到未预热的生产者，链路时间增加了连接时间，RT时间必然会增加，从而也会导致DUBBO线程池打满问题。又加上查询超时重试。
### 解决方案
(1) 等待生产者充分预热
因为生产者预热不充分导致线程池打满问题，最容易发生在系统发布时。例如发布了一台机器后发现线上出现线程池打满问题，千万不要着急重启机器，而是给机器一段时间预热，等连接建立后问题大概率消失。同时我们在发布时也要分多批次发布，不要一次发布太多机器导致服务因为预热问题造成大面积影响。
(2) DUBBO升级版本大于等于2.7.4
DUBBO消费者在调用选择生产者时本身就会执行预热逻辑，为什么还会出现预热不充分问题？这是因为2.5.5之前版本以及2.7.2版本预热机制是有问题的，简而言之就是获取启动时间不正确，2.7.4版本彻底解决了这个问题，所以我们要避免使用问题版本。下面我们阅读2.7.0版本预热机制源码，看看预热机制如何生效：
```java
public class RandomLoadBalance extends AbstractLoadBalance {
public static final String NAME = "random";
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// invokers数量
int length = invokers.size();
// 权重是否相同
boolean sameWeight = true;
// invokers权重数组
int[] weights = new int[length];
// 第一个invoker权重
int firstWeight = getWeight(invokers.get(0), invocation);
weights[0] = firstWeight;
// 权重值之和
int totalWeight = firstWeight;
for (int i = 1; i < length; i++) {
// 计算权重值
int weight = getWeight(invokers.get(i), invocation);
weights[i] = weight;
totalWeight += weight;
// 任意一个invoker权重值不等于第一个invoker权重值则sameWeight设置为FALSE
if (sameWeight && weight != firstWeight) {
sameWeight = false;
}
}
// 权重值不等则根据总权重值计算
if (totalWeight > 0 && !sameWeight) {
int offset = ThreadLocalRandom.current().nextInt(totalWeight);
// 不断减去权重值当小于0时直接返回
for (int i = 0; i < length; i++) {
offset -= weights[i];
if (offset < 0) {
return invokers.get(i);
}
}
}
// 所有服务权重值一致则随机返回
return invokers.get(ThreadLocalRandom.current().nextInt(length));
}
}
public abstract class AbstractLoadBalance implements LoadBalance {
static int calculateWarmupWeight(int uptime, int warmup, int weight) {
// uptime/(warmup*weight)
// 如果当前服务提供者没过预热期，用户设置的权重将通过uptime/warmup减小
// 如果服务提供者设置权重很大但是还没过预热时间，重新计算权重会很小
int ww = (int) ((float) uptime / ((float) warmup / (float) weight));
return ww < 1 ? 1 : (ww > weight ? weight : ww);
}
protected int getWeight(Invoker<?> invoker, Invocation invocation) {
// 获取invoker设置权重值默认权重=100
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
// 如果权重大于0
if (weight > 0) {
// 服务提供者发布服务时间戳
long timestamp = invoker.getUrl().getParameter(Constants.REMOTE_TIMESTAMP_KEY, 0L);
if (timestamp > 0L) {
// 服务已经发布多少时间
int uptime = (int) (System.currentTimeMillis() - timestamp);
// 预热时间默认10分钟
int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);
// 生产者发布时间大于0但是小于预热时间
if (uptime > 0 && uptime < warmup) {
// 重新计算权重值
weight = calculateWarmupWeight(uptime, warmup, weight);
}
}
}
// 服务发布时间大于预热时间直接返回设置权重值
return weight >= 0 ? weight : 0;
}
}
```
(3) 找出慢服务
DUBBO线程池打满时会执行拒绝策略：
```java
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();
// 每10分钟输出线程快照
if (now - lastPrintTime < 10 * 60 * 1000) {
return;
}
if (!guard.tryAcquire()) {
return;
}
ExecutorService pool = Executors.newSingleThreadExecutor();
pool.execute(() -> {
String dumpPath = url.getParameter(Constants.DUMP_DIRECTORY, System.getProperty("user.home"));
System.out.println("AbortPolicyWithReport dumpJStack directory=" + dumpPath);
SimpleDateFormat sdf;
String os = System.getProperty("os.name").toLowerCase();
// linux文件位置/home/xxx/Dubbo_JStack.log.2021-01-01_20:50:15
// windows文件位置/user/xxx/Dubbo_JStack.log.2020-01-01_20-50-15
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());
try (FileOutputStream jStackStream = new FileOutputStream(new File(dumpPath, "Dubbo_JStack.log" + "." + dateStr))) {
JVMUtil.jstack(jStackStream);
} catch (Throwable t) {
logger.error("dump jStack error", t);
} finally {
guard.release();
}
lastPrintTime = System.currentTimeMillis();
});
pool.shutdown();
}
}
```
拒绝策略会输出线程快照文件，在分析线程快照文件时BLOCKED和TIMED_WAITING线程状态需要我们重点关注。如果发现大量线程阻塞或者等待状态则可以定位到具体代码行：
```java
ubboServerHandler-x.x.x.x:9999-thread-200 Id=230 TIMED_WAITING
at java.lang.Thread.sleep(Native Method)
at com.java.front.dubbo.demo.provider.HelloServiceImpl.sayHello(HelloServiceImpl.java:13)
at org.apache.dubbo.common.bytecode.Wrapper1.invokeMethod(Wrapper1.java)
at org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory$1.doInvoke(JavassistProxyFactory.java:56)
at org.apache.dubbo.rpc.proxy.AbstractProxyInvoker.invoke(AbstractProxyInvoker.java:85)
at org.apache.dubbo.config.invoker.DelegateProviderMetaDataInvoker.invoke(DelegateProviderMetaDataInvoker.java:56)
at org.apache.dubbo.rpc.protocol.InvokerWrapper.invoke(InvokerWrapper.java:56)
```
公众号进入网站体验更佳
