记一次Netty连接池FixedChannelPool连接未释放问题的排查总结
1 前言
前几天我们又遇到了一个Netty报从连接池获取连接超时异常从而导致整个服务不可用的异常，报的具体异常信息是Exception accurred when acquire channel channel pool:TimeoutException。当时自己看了这个异常信息，有种似曾相识的感觉，印象中自己第一次接触到该异常是不久前也遇到了Netty报超时错误导致整个服务不可用的问题，最终只能重启服务器来解决。于是自己去翻看了之前的异常消息，发现报的错误果真同样是从连接池获取连接超时的异常！印象中前段时间Netty报这个错误时是刚好相关网络部门做过网络调整，当时我们就认为可能是由于网络原因导致Netty获取连接超时，但是至于为啥会因为网络原因导致获取Netty连接超时后从而导致服务不可用就还是一无所知，因此，这个“幽灵”Bug暂时对我们来说成了一团谜。
2 “幽灵”Bug得以复现给了我们解决这个Bug的希望
万幸的是，这次相关同事复现了这个Bug，然后对方说只要在并发量大一点且后台业务逻辑处理时间久的话这个Bug就会复现，且这个Bug是伴随前台线程请求后台超时（这个是请求超时异常，而非获取连接超时异常，注意区分）后报出来的。于是自己提高并发量且在后台模拟业务超时进行测试，果真“幽灵”Bug得以复现了，且这个Bug导致后面整个服务都不可用了，报错如下截图：
这个“幽灵”Bug的复现给我们带来了解决它的希望，那么是什么原因导致在并发量一上来且前台请求后台超时后就会导致从Netty连接池获取连接超时了呢？
注意： 这里有两个超时异常，请注意区分：一个是从连接池获取连接超时异常；令一个是从连接池成功获取连接后，前台请求后台，由于后台业务逻辑执行时间过长导致抛出的请求超时异常
我们无从而知，只能去翻看抛异常的代码，我们编写的Netty连接池实现大概如下：
// CustomChannelPool.java
public class CustomChannelPool {
private ChannelHealthChecker healthCheck = ChannelHealthChecker.ACTIVE;
acquireTimeoutAction = null;
acquireTimeoutMillis = -1;
maxConnect = 8;
maxPendingAcquires = Integer.MAX_VALUE
releaseHealthCheck=true
//...省略无关属性
static ChannelPool fixpool =
new FixedChannelPool(b, handler, healthCheck, acquireTimeoutAction,
acquireTimeoutMillis, maxConnect, maxPendingAcquires, releaseHealthCheck, lastRecentUsed); // 【0】
// 获取连接
public Channel acquire(int timeoutMillis) throws Exception {
try {
Future<Channel> fch = fixpool.acquire(); // 【1】
Channel ch = fch.get(timeoutMillis, TimeUnit.MILLISECONDS);// 【2】
return ch;
} catch (Exception e) {
logger.error("Exception accurred when acquire channel from channel pool.", e);//【3】
throw e; //【4】
}
}
// 释放连接
public void release(Channel channel) {
try {
if (channel != null) {
fixpool.release(channel);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
然后业务获取连接的代码大概如下：
// BusineseService.java
public class BusineseService {
public Response rpcCall() throw Exception{
// 获取连接
Channel channel = CustomChannelPool.fixpool.acquire(10000); // 【5】
try {
// ...省略相关业务逻辑
// 最终进行底层远程调用
channel.writeAndFlush(data);
// ...省略相关业务逻辑
} finally {
// 释放连接
// 若前台请求后台超时后，是有释放连接的
CustomChannelPool.fixpool.release(channel); // 【6】
}
}
}
根据报的异常信息可用判断是在进行远程调用前调用CustomChannelPool.acquire方法的Channel ch = fch.get(timeoutMillis, TimeUnit.MILLISECONDS);这句代码从Netty连接池获取连接超时（即10秒后），然后抛出TimeoutException，最后再在CustomChannelPool.acquire方法的catch代码块打印出Exception accurred when acquire channel from channel pool:TimeoutException异常信息，然后再把该异常往外抛出去,即最后会在BusineseService的标号【5】处的代码Channel channel = CustomChannelPool.fixpool.acquire(10000);抛出了一个TimeoutException异常，又因为标号【5】处的代码没有包含在try块内，因此不会执行标号【6】处的finally块释放连接的逻辑。
分析到这里，我们松了一口气，原来导致该“幽灵”Bug的原因就是因为获取连接的这决代码Channel channel = CustomChannelPool.fixpool.acquire(10000);没有被try块包围住，才导致没有执行finlly块的释放连接逻辑！！！
Please calm down here!
即使我们将获取连接的这决代码Channel channel = CustomChannelPool.fixpool.acquire(10000);用try块包围住，最终在执行finally块释放连接的逻辑时等待我们的将会是什么呢？显然，等待我们的是一个空指针异常！为啥呢？因为执行Channel channel = CustomChannelPool.fixpool.acquire(10000);这句代码抛出TimeoutException异常后，拿到的channel将为null，然后我们再用结果为null的channel去释放连接，自然会抛出一个NPE.
之前燃起的一线希望又被NPE扑灭了，出现幽灵Bug的原因依然没找到！此时我们又失去了方向！
既然选择了远方,便只顾风雨兼程，嘿嘿，这里我们自我励志下，别灰心，努力了总能解决它，不就是一个小小的bug么。
于是我们又冷静分析了下出问题的两句问题代码上来：
Future<Channel> fch = fixpool.acquire(); // 【1】
Channel ch = fch.get(timeoutMillis, TimeUnit.MILLISECONDS);// 【2】
【1】处代码调用fixpool.acquire()方法去获取一个连接然后马上返回一个Future<Channel>对象fch，紧接着我们再调用【2】处代码fch.get(timeoutMillis, TimeUnit.MILLISECONDS);方法来等待连接池的可用连接返回，一直阻塞直至超时，超时后就抛出了TimeoutException异常。
从这里初步分析可以看到Netty获取连接是异步进行的，当获取到一个连接后再唤醒调用fch.get(timeoutMillis, TimeUnit.MILLISECONDS);代码后正在阻塞等待的线程。
我们再回想下，复现该bug的前提条件是抛出该异常的前提是并发量大且会伴随着大量前台请求后台的线程请求超时后出现。这里请求后台超时的线程是已经成功从连接池获得连接的线程，且超时抛出请求超时异常后也有执行finally块的释放（归还）连接回连接池的操作的！
那么是什么原因会导致抛出从连接池获取连接超时异常呢？于是我们不禁有以下猜测：
猜测1： 瞬间高并发的请求导致连接池资源耗尽，从而导致大量获取连接超时，这种情况是可能出现的，但是高并发过后，整个服务就不可用了（这里的服务不可用不是指应用宕掉，而是总是报获取连接超时）！按理说高并发过后应该归还连接到连接池了，因此肯定不会出现服务不可用的情况。因此这个猜测可以排除了，唯一的原因就是连接没能正常归还到连接池！！！
至于为啥连接没能正常归还到连接池，我们又有以下猜测：
猜测2： 请求后台超时的channel连接不能正常归还到连接池，channel连接请求后台超时后，这个连接不能正常放回连接池，导致channel连接池可用连接耗尽，最终导致其他线程从连接池获取连接超时？如果是这样，那么为啥请求后台超时的连接不能正常放回连接池呢？
猜测3： 请求后台超时channel连接能正常归还到连接池，此时又因为从连接池获取channel连接是异步的，当获取连接超时后，我们关心的是获取连接的异步线程最终有无从连接池成功获取到一个连接呢？这里有两种可能：1）获取连接超时后不能从连接池获取到一个连接，即使前面实现的代码中获取连接超时的话没有释放连接也不影响，因为这种情况根本就没有获取到连接；1）获取连接超时后仍能成功获取到一个连接，但从前面实现代码的分析过程中可以知道，获取连接超时的话，这个获取到的连接是没有被释放的，如果是这种情况，那么就会导致连接池资源耗尽从而导致服务不可用了！
显然，我们要朝着猜测2和猜测3的方向去排查问题，至于哪种原因导致连接没能正常归还到连接池呢？我们依然百思不得其解！因为此时Netty连接池对于我们来说是一个黑盒，此时是时候去打开这个黑盒一探究竟了！
3 Netty连接池FixedChannelPool获取和释放连接源码分析
来到这里我们就要打开Netty的channel连接池源码看一下了，前面导致问题的代码无非就是连接池的acquire和release两个方法，相信我们能从连接池的这两个方法的源码中找到导致Exception accurred when acquire channel from channel pool:TimeoutException异常即获取连接超时异常的原因。
3.1 连接池整体类结构的理解
这里用到的是Netty的FixedChannelPool连接池，同时FixedChannelPool继承了SimpleChannelPool，而SimpleChannelPool又实现了ChannelPool接口，如下图：
我们先来看下ChannelPool接口的源码：
// ChannelPool.java
public interface ChannelPool extends Closeable {
Future<Channel> acquire();
Future<Channel> acquire(Promise<Channel> promise);
Future<Void> release(Channel channel);
Future<Void> release(Channel channel, Promise<Void> promise);
void close();
}
可见ChannelPool接口实现了Netty连接池获取连接和释放连接的基本接口，而相应的获取连接和释放连接的返回结果类时都是Future类型，可见Netty连接池获取连接和释放连接的操作都是异步执行的。
源码这里先补贴了，我们来看下SimpleChannelPool的类结构：
首先SimpleChannelPool实现了Netty的channel连接池的基本功能如获取连接，释放连接以及对channel连接进行健康检查等。此外，SimpleChannelPool是如何来存储channel连接呢？此时从上图的序号4可以看到定义了一个双端队列deque来存储channel连接。
再来看下FixedChannelPool的类结构：
可以看到FixedChannelPool在SimpleChannelPool的基础上实现了连接池数量控制，待获取连接超时任务处理，待获取连接超时任务处理策略以及释放连接后唤醒待获取连接的任务的一些逻辑。详细解析如下：
- 成员变量
maxPendingAcquires表示连接池的最大连接数即连接池容量，pendingAcquireCount表示已经获取的连接数量（包括从连接池建立的连接及额外新建的连接）。这两个变量用来判断连接池有无可用连接； - 内部类
AcquireTask,待获取连接任务，当连接池资源耗尽时，待获取的连接会被封装成一个AcquireTask任务； - 定义了一个
ArrayDeque类型的双端队列pendingAcquireQueue，当连接池可用channel连接耗尽时，待获取的连接会被封装成一个AcquireTask，然后pendingAcquireQueue队列就是用来存储AcquireTask的； - 成员变量
maxPendingAcquires表示pendingAcquireQueue队列的大小，pendingAcquireCount表示等待获取channel连接的数量，这两个变量用来控制pendingAcquireQueue队列容量满还是不满； - 成员变量
acquireTimeoutNanos表示从连接池获取channel连接的超时时间，内部枚举类AcquireTimeoutAction封装了待获取连接的任务超时时该执行的策略，默认有新建NEW和失败FAIL策略； - 内部抽象类
TimeoutTask实现了Runnable接口，当待获取连接任务超时时，此时根据AcquireTimeoutAction策略来执行该任务；
3.2 从连接池获取连接的源码分析
我们首先来分析下连接池获取连接的源码，直接上源码：
// FixedChannelPool.java
@Override
public Future<Channel> acquire(final Promise<Channel> promise) {
try {
// 如果当前线程是executor的线程，那么就直接调用acquire0方法获取连接，
// 【注意】这里是异步去获取channel连接哈，如果调用future.get方法，只要连接没获取到，那么将一直阻塞，直到连接获取完成。
if (executor.inEventLoop()) {
acquire0(promise);
// 如果当前线程不是executor的线程，那么就由executor这个线程调用acquire0方法获取连接,这里是异步获取连接哈
} else {
executor.execute(new Runnable() {
@Override
public void run() {
acquire0(promise);
}
});
}
} catch (Throwable cause) {
// 出现异常，设置失败回调
promise.setFailure(cause);
}
// 返回保证，这里的保证是能拿到Channel，Promise继承了Future
re
可以看到acquire方法又调用了acquire0方法：
// FixedChannelPool.java
private void acquire0(final Promise<Channel> promise) {
assert executor.inEventLoop();
// 判断FixedChannelPool连接池是否已经关闭
if (closed) {
promise.setFailure(new IllegalStateException("FixedChannelPool was closed"));
return;
}
// 【1】如果已经获取的连接数量acquiredChannelCount小于Channel连接池的数量，说明连接池还有可用连接，因此这里直接从池子里取连接即可
// 注意：acquiredChannelCount是从0开始计数的哈
if (acquiredChannelCount.get() < maxConnections) {
assert acquiredChannelCount.get() >= 0;
// We need to create a new promise as we need to ensure the AcquireListener runs in the correct
// EventLoop
Promise<Channel> p = executor.newPromise();
// 新建一个AcquireListener，这个AcquireListener是FixedChannelPool的一个内部类，
// TODO 用来当获取到连接回调其内部的operationComplete方法？
AcquireListener l = new AcquireListener(promise);
// 调用AcquireListener的acquired方法，在获取到连接前先给acquiredChannelCount加1，
// TODO [思考]大胆猜测，如果后续的获取连接若失败，肯定有acquiredChannelCount减1的代码，但是在哪里呢
l.acquired();
// 给保证添加AcquireListener监听器
p.addListener(l);
// 这里还是调用父类SimpleChannelPool来获取连接，这里先提下父类SimpleChannelPool没有实现连接池数量控制的相关功能，
// SimpleChannelPool只是实现了新建连接，健康检查等逻辑
super.acquire(p);
// 【2】获取连接时，能执行到这里，说明已经获取的连接数量acquiredChannelCount大于或等于Channel连接池的数量，
// 即表明连接池无可用连接了，此时就需要根据有无设置AcquireTimeoutAction策略来执行相应的操作了
} else {
// 【2.1】如果等待获取连接数量pendingAcquireCount超过队列的最大容量maxPendingAcquires的话，此时直接抛异常
if (pendingAcquireCount >= maxPendingAcquires) {
tooManyOutstanding(promise);
// 【2.2】若等待获取连接数量pendingAcquireCount还没占满pendingAcquireQueue队列
} else {
// 这里把等待获取连接的保证promise封装成AcquireTask任务
AcquireTask task = new AcquireTask(promise);
// 将之前封装的AcquireTask任务入pendingAcquireQueue队列，放到最后面,这里pendingAcquireQueue是一个ArrayDeque队列
if (pendingAcquireQueue.offer(task)) {
// 入队成功，因此pendingAcquireCount自增1
++pendingAcquireCount;
// 【重要】这里如果timeoutTask不为null，则说明要么设置了获取连接超时的处理策略，目前的netty连接池内置的策略中，要么为NEW，要么为FAIL
if (timeoutTask != null) {
// 设置了获取连接超时处理策略的话，那么把timeoutTask扔到定时任务里去，一旦获取连接超时，那么就执行timeoutTask
// 若策略为NEW，那么就会新建连接然后返回；若策略为FAIL，那么直接抛异常
// 这里调度超时任务后，然后再给task.timeoutFuture赋值，也是为了做标记的意思，因为后面一个线程释放连接后
// 会继续“唤醒”pendingAcquireQueue的一个任务，那时候这个任务肯定是未超时的，所以需要取消这个定时任务
task.timeoutFuture = executor.schedule(timeoutTask, acquireTimeoutNanos, TimeUnit.NANOSECONDS);
}
// 执行到这里，说明前面入pendingAcquireQueue队列时队列已满，然后也直接抛异常
} else {
tooManyOutstanding(promise);
}
}
assert pendingAcquireCount > 0;
}
}
这里再着重看下当带获取连接超时后，这个TimeoutTask的源码逻辑是怎样的：
// FixedChannelPool.TimeoutTask.java
private abstract class TimeoutTask implements Runnable {
@Override
public final void run() {
assert executor.inEventLoop();
// 获取系统当前时间
long nanoTime = System.nanoTime();
// 进入死循环
for (;;) {
// 从pendingAcquireQueue队列中获取一个待获取连接的任务，【注意】这里是peek哈，相当于查询，而不会移除队列中的元素
// 【思考】天哪，这里是死循环+查询队列的操作，那当一个获取连接超时定时任务到来时，岂不会将pendingAcquireQueue队列中的
// 所有任务（包括未timeout的任务）都查出来？可以肯定的是这里确实是这样子，答案见后面代码注释分析
AcquireTask task = pendingAcquireQueue.peek();
// Compare nanoTime as descripted in the javadocs of System.nanoTime()
//
// See https://docs.oracle.com/javase/7/docs/api/java/lang/System.html#nanoTime()
// See https://github.com/netty/netty/issues/3705 这里估计出现过bug，后面修复了，嘿嘿。以后有空再去看看这个issue
// （1）如果从pendingAcquireQueue队列获取的任务为空，那么则说明没有待获取连接的任务了，此时直接break；
// (2) 如果从pendingAcquireQueue队列获取的任务不为空，此时肯定不能直接进行remove操作吧，想想，此时pendingAcquireQueue队列里
// 是不是有可能还有未超时的任务，
// 2.1）因此下面需要执行nanoTime - task.expireNanoTime是不是小于0，如果小于0直接break，等这个任务超时时再来执行这里的代码，
// 想想如果这里将非超时的任务也一起取出来去执行也不是不可以，想想这里不这样做的原因如下：
// a)这里的职责是专门处理获取连接超时任务的，如果这里也执行非超时任务，那么造成功能混乱；
// b)若本来超时任务就多，此时又加上处理非超时任务的话，那么系统压力会更大
// c)这里非超时任务应该留给连接池的可用连接去处理哈，因为这里pendingAcquireQueue里的任务本来就是因为连接池资源耗尽的情况下，
// 其余获取连接的任务才入pendingAcquireQueue队列的，因此当一个线程用完从连接池获取的连接后，这个线程把连接归还给连接池后，
// 这个线程首先判断连接池还有无可用连接，若连接池还有可用连接，那么其有义务有“唤醒”pendingAcquireQueue队列中的一个未超时的任务，
// 这个任务被唤醒后，然后再去连接池获取连接即可
// 2.2）如果大于等于0，那么就根据是NEW还是FAIL策略来执行这个获取连接超时任务了
if (task == null || nanoTime - task.expireNanoTime < 0) {
break;
}
// 执行到这里，说明获取连接任务确实超时了，因此可以将这个任务直接从pendingAcquireQueue队列移除了哈
pendingAcquireQueue.remove();
// 自然，pendingAcquireCount也会减1
--pendingAcquireCount;
// 还记得FixedChannelPool的一个构造方法最终会根据AcquireTimeoutAction的NEW还是FAIL策略来新建一个TimeoutTask，
// 然后当获取连接时连接池又无可用连接情况下，此时除了获取连接任务会入pendingAcquireQueue队列外，另外TimeoutTask也会交给
// 一个定时任务调度线程线程去执行，还记得么？
// 那么代码执行到这里，说明已经在这个定时任务的调度方面里面了，此时再回调TimeoutTask的onTimeout方法哈
onTimeout(task);
}
}
// 根据带获取连接任务超时时，该回调的策略方法
public abstract void onTimeout(AcquireTask task);
}
同样，详情请见注释即可，我们再来看下根据带获取连接超时后，最后会执行回调onTimeout方法，那么我们再来看看onTimeout方法的相关逻辑：
// FixedChannelPool.java
public FixedChannelPool(Bootstrap bootstrap,
ChannelPoolHandler handler,
ChannelHealthChecker healthCheck, AcquireTimeoutAction action,
final long acquireTimeoutMillis,
int maxConnections, int maxPendingAcquires,
boolean releaseHealthCheck, boolean lastRecentUsed) {
super(bootstrap, handler, healthCheck, releaseHealthCheck, lastRecentUsed);
if (maxConnections < 1) {
throw new IllegalArgumentException("maxConnections: " + maxConnections + " (expected: >= 1)");
}
if (maxPendingAcquires < 1) {
throw new IllegalArgumentException("maxPendingAcquires: " + maxPendingAcquires + " (expected: >= 1)");
}
// 这里表示初始化时获取连接超时action策略为null且acquireTimeoutMillis == -1
if (action == null && acquireTimeoutMillis == -1) {
timeoutTask = null;
acquireTimeoutNanos = -1;
// 做一些不合理的参数校验
} else if (action == null && acquireTimeoutMillis != -1) {
throw new NullPointerException("action");
// 做一些不合理的参数校验
} else if (action != null && acquireTimeoutMillis < 0) {
throw new IllegalArgumentException("acquireTimeoutMillis: " + acquireTimeoutMillis + " (expected: >= 0)");
// 执行到这里，表示action != null且acquireTimeoutMillis >= -1，即设置了获取连接超时的从处理策略
} else {
acquireTimeoutNanos = TimeUnit.MILLISECONDS.toNanos(acquireTimeoutMillis);
// 判断是NEW还是FAIL策略
switch (action) {
// （1）如果是获取连接超时FAIL策略，当获取连接超时的话，此时如果pendingAcquireQueue队列中还有未能拿到连接的线程任务，此时直接失败抛异常，简单粗暴！！！
case FAIL:
timeoutTask = new TimeoutTask() {
@Override
public void onTimeout(AcquireTask task) {
// Fail the promise as we timed out.
task.promise.setFailure(new TimeoutException(
"Acquire operation took longer then configured maximum time") {
@Override
public Throwable fillInStackTrace() {
return this;
}
});
}
};
break;
// （2）如果是获取连接超时NEW策略，当获取连接超时的话，此时如果pendingAcquireQueue队列中还有未能拿到连接的线程任务，
// 此时会为这些获取连接的线程任务新建连接，这里理性一点，到时如果设置pendingAcquireCount过大，在高并发情况下会导致大量连接创建
// 有着耗尽资源的风险
case NEW:
timeoutTask = new TimeoutTask() {
@Override
public void onTimeout(AcquireTask task) {
// Increment the acquire count and delegate to super to actually acquire a Channel which will
// create a new connection.
// acquiredChannelCount获取的连接数+1且给acquired赋值true
task.acquired();
// 调用父类SimpleChannelPool.acquire来创建一直新连接
FixedChannelPool.super.acquire(task.promise);
}
};
break;
default:
throw new Error();
}
}
// 这个executor是用来获取连接的，总是同一个executor异步去获取连接
executor = bootstrap.config().group().next();
this.maxConnections = maxConnections;
this.maxPendingAcquires = maxPendingAcquires;
}
可见，原来TimeoutTask.onTimeout方法是在FixedChannelPool的构造方法中初始化的即当我们新建一个Netty连接池FixedChannelPool时TimeoutTask.onTimeout方法就会根据超时任务策略初始化好，详情见源码注释即可。
3.3 释放（归还）连接回连接池的源码分析
前面分析了Netty连接池FixedChannelPool获取连接的过程，下面我们同样来分析下Netty连接池FixedChannelPool释放连接的源码，因为释放连接是直接调用了父类SimpleChannelPool的release方法：
// SimpleChannelPool.java
@Override
public final Future<Void> release(Channel channel) {
// 这里如果连接池是FixedChannelPool的话，这里实质调用的是FixedChannelPool的release(final Channel channel, final Promise<Void> promise)方法，
// 因为FixedChannelPool重载了SimpleChannelPool的release(final Channel channel, final Promise<Void> promise)方法
return release(channel, channel.eventLoop().<Void>newPromise());
}
此时在SimpleChannelPool的release方法中又调用了子类FixedChannelPool的重载的release(channel, channel.eventLoop().<Void>newPromise());方法，我们进该方法一看究竟：
// FixedChannelPool.java
@Override
public Future<Void> release(final Channel channel, final Promise<Void> promise) {
ObjectUtil.checkNotNull(promise, "promise");
// 新建一个Promise
final Promise<Void> p = executor.newPromise();
// 然后再调用父类SimpleChannelPool的release(final Channel channel, final Promise<Void> promise)方法，
// 【思考】这里为啥要这么绕呀？？？先是调用父类SimpleChannelPool的release(Channel channel)，然后在父类SimpleChannelPool的release方法
// 中再调用本方法，而明明父类就有这个release(final Channel channel, final Promise<Void> promise)方法，为何不直接调用呢？？？
//【答案】答案就是SimpleChannelPool只实现了连接池获取连接，释放连接和健康检查的相关基本方法，而连接释放回连接池后，我们是不是要唤醒
// pendingAcquireQueue队列中的一个任务呢？是吧，因此下面就给Promise又添加了一个FutureListener监听器，这个监听器的作用就是当SimpleChannelPool的
// release方法把连接放回连接池后，此时回调该监听器的operationComplete方法来唤醒pendingAcquireQueue里的一个任务，嘿嘿，是不是有点绕，哈哈
super.release(channel, p.addListener(new FutureListener<Void>() {
@Override
public void operationComplete(Future<Void> future) throws Exception {
assert executor.inEventLoop();
// 以为连接池已经关闭，我们没得选择只能关闭channel，然后回调setFailure反弹广发
if (closed) {
// Since the pool is closed, we have no choice but to close the channel
channel.close();
promise.setFailure(new IllegalStateException("FixedChannelPool was closed"));
return;
}
// （1）如果释放连接回连接池成功
// TODO【思考】这个future是只哪个future呢？你能找到这个future是从哪里传进来的吗？嘿嘿嘿
if (future.isSuccess()) {
// 那么此时就要就获取的连接数量acquiredChannelCount减1且“唤醒”pendingAcquireQueue队列的一个待获取连接的一个任务
// 还记得之前分析acquire源码时当连接池无可用连接时，此时会将这个获取连接的一个线程封装成一个AcquireTask任务放进pendingAcquireQueue队列吗？
decrementAndRunTaskQueue();
// 回到setSuccess方法
promise.setSuccess(null);
// （2）如果连接没有成功释放回连接池，且没有还错池子的情况下发生了异常，那么这里同样需要获取的连接数量acquiredChannelCount减1
// 且“唤醒”pendingAcquireQueue队列的一个待获取连接的一个任务
// TODO 纳尼？？这里还可以多个池子？为啥没还错池子的情况发生了归还连接的时候发生异常就不用 decrementAndRunTaskQueue呢？二十直接调用setFailure方法，
// 这个setFailure方法又因此着什么逻辑呢？
} else {
Throwable cause = future.cause();
// Check if the exception was not because of we passed the Channel to the wrong pool.
if (!(cause instanceof IllegalArgumentException)) {
decrementAndRunTaskQueue();
}
// 回调setFailure方法
promise.setFailure(future.cause());
}
}
}));
return promise;
}
从源码可以看到调用了子类FixedChannelPool的重载的release(channel, channel.eventLoop().<Void>newPromise());方法是为了添加一个FutureListener监听器，这个监听器的作用详见注释，然后又调回父类SimpleChannelPool的release(final Channel channel, final Promise<Void> promise)方法，是不是有点绕，嘿嘿。继续看该方法源码：
// SimpleChannelPool.java
public Future<Void> release(final Channel channel, final Promise<Void> promise) {
checkNotNull(channel, "channel");
checkNotNull(promise, "promise");
try {
// TODO 【思考】这里每次释放连接都是有多个NioEventLoop线程，而获取连接却用的是同一个NioEventLoop线程，为啥？？？
EventLoop loop = channel.eventLoop();
// TODO 【思考】这里何时会被执行到？经过调试一般都是执行的是else分支
if (loop.inEventLoop()) {
doReleaseChannel(channel, promise);
} else {
// 此时将释放连接的操作封装成一个Runnable任务，然后将这个任务添加进SingleThreadEventExecutor的taskQueue中
// 反正最终是异步释放连接
loop.execute(new Runnable() {
@Override
public void run() {
doReleaseChannel(channel, promise);
}
});
}
} catch (Throwable cause) {
closeAndFail(channel, cause, promise);
}
return promise;
}
结果父类SimpleChannelPool的release方法中又继续调用doReleaseChannel方法来释放连接，由于篇幅有限，这里更具体的源码就不再深究了。不过可以肯定的是调用完doReleaseChannel方法释放连接后，必然会回调之前添加的FutureListener的operationComplete方法，然后继续调用decrementAndRunTaskQueue方法，那么我们继续跟下decrementAndRunTaskQueue方法源码：
// FixedChannelPool.java
private void decrementAndRunTaskQueue() {
// We should never have a negative value.
// 因为前面已经把连接归还回连接池了，自然这里会将已获取的连接数量减1
int currentCount = acquiredChannelCount.decrementAndGet();
assert currentCount >= 0;
// Run the pending acquire tasks before notify the original promise so if the user would
// try to acquire again from the ChannelFutureListener and the pendingAcquireCount is >=
// maxPendingAcquires we may be able to run some pending tasks first and so allow to add
// more.
// 然后“唤醒”pendingAcquireQueue队列的一个待获取连接的一个任务去连接池拿连接，
// 因为这里唤醒的是未超时的任务，因此连接必须从连接池拿
runTaskQueue();
}
继续跟runTaskQueue方法源码：
// FixedChannelPool.java
private void runTaskQueue() {
// 这里非超时任务应该留给连接池的可用连接去处理哈，因为这里pendingAcquireQueue里的任务本来就是因为连接池资源耗尽的情况下，
// 其余获取连接的任务才入pendingAcquireQueue队列的，因此当一个线程用完从连接池获取的连接后，这个线程把连接归还给连接池后，
// 这个线程首先判断连接池还有无可用连接，若连接池还有可用连接，那么其有义务有“唤醒”pendingAcquireQueue队列中的一个未超时的任务，
// 这个任务被唤醒后，然后再去连接池获取连接即可
// 如果acquiredChannelCount小于连接池数量，说明连接池还有可用连接
// TODO 【思考】这里while (acquiredChannelCount.get() < maxConnections)判断的初衷感觉像是一定要从连接池获取一个连接，
// 而不是新建一个连接，否则就不用这么判断了。那么问题来了：
// while (acquiredChannelCount.get() < maxConnections)没有线程安全问题么？？？如果不用锁的话可能会出现“一票多卖”问题
// 除非这里是单线程执行就没有线程安全问题。
// 如果存在线程安全问题，当并发量大的话出现“一票多卖问题”，即最终还会导致连接池可用连接耗尽，其他没能拿到连接的线程还是会新建
// 一些连接出来，这么做可是可以，但却又违反了“未超时任务的连接只能等待线程池的连接，超时任务再由定时任务额外新建连接”的初衷，
// 因为执行到这里从pendingAcquireQueue队列取出的任务的一般都是未超时的。
// 答案这里应该是单线程执行？待确认？调试的时候发现基本是同一个线程
//
while (acquiredChannelCount.get() < maxConnections) {
// 取出第一个待获取连接的未超时的任务，因为如果是超时的获取连接任务的话，已经被定时任务移除掉了哈
AcquireTask task = pendingAcquireQueue.poll();
// 若队列里没有待获取连接的任务，直接跳出即可
if (task == null) {
break;
}
// 如果当初有设置定时任务清理超时的带获取连接任务，那么此时timeoutFuture不为Null，因此需要取消这个定时任务的执行
// Cancel the timeout if one was scheduled
ScheduledFuture<?> timeoutFuture = task.timeoutFuture;
if (timeoutFuture != null) {
timeoutFuture.cancel(false);
}
// pendingAcquireCount减1
--pendingAcquireCount;
// acquiredChannelCount加1
task.acquired();
// 调用父类SimpleChannelPool的acquire方法：
// 1）连接池有可用连接，从连接池取出即可；
// 2）连接池没有可用连接，此时直接NEW一个连接出来
super.acquire(task.promise);
}
// We should never have a negative value.
assert pendingAcquireCount >= 0;
assert acquiredChannelCount.get() >= 0;
}
可见，当获取到连接的线程将连接放回连接池后，会继续唤醒一些pendingAcquireQueue队列未超时的待获取连接的任务来获取连接。下面继续用一个流程图来总结下释放连接的过程：
3.4 Netty连接池获取和释放连接流程总结
同样，以一个流程图来总结Netty连接池获取和释放连接流程:
这里不再文字累赘总结，更详细的Netty源码注释可参见我的github网址：https://github.com/yuanmabiji/netty
分析完 Netty连接池获取和释放连接流程，前面的猜测2和猜测3全部可以得到答案了：
猜测2： 请求后台超时的channel连接不能正常归还到连接池，channel连接请求后台超时后，这个连接不能正常放回连接池，导致channel连接池可用连接耗尽，最终导致其他线程从连接池获取连接超时？如果是这样，那么为啥请求后台超时的连接不能正常放回连接池呢？
猜测2答案： 请求后台超时的channel连接可以正常释放回连接池，且放回的连接是健康可用的。
猜测3： 请求后台超时channel连接能正常归还到连接池，此时又因为从连接池获取channel连接是异步的，当获取连接超时后，我们关心的是获取连接的异步线程最终有无从连接池成功获取到一个连接呢？这里有两种可能：1）获取连接超时后不能从连接池获取到一个连接，即使前面实现的代码中获取连接超时的话没有释放连接也不影响，因为这种情况根本就没有获取到连接；1）获取连接超时后仍能成功获取到一个连接，但从前面实现代码的分析过程中可以知道，获取连接超时的话，这个获取到的连接是没有被释放的，如果是这种情况，那么就会导致连接池资源耗尽从而导致服务不可用了！
猜测3答案： 获取连接超时后仍能成功获取到一个连接，但从前面实现代码的分析过程中可以知道，获取连接超时的话，这个获取到的连接是没有被释放的，如果是这种情况，那么就会导致连接池资源耗尽从而导致服务不可用了！
4 获取连接超时异常导致连接池资源耗尽的Bug原因分析
前面详细分析了Netty连接池获取连接和释放连接的流程，相信一直困扰着我们的“幽灵”Bug的原因已经付出水面了吧。这里直接开门见山说答案吧，现在还是先来回复下我们的问题代码：
// CustomChannelPool.java
public class CustomChannelPool {
private ChannelHealthChecker healthCheck = ChannelHealthChecker.ACTIVE;
// 【重要，问题代码】acquireTimeoutAction策略居然是null！
acquireTimeoutAction = null;
// 【重要，问题代码】acquireTimeoutAction策略是null，且这里acquireTimeoutMillis = -1
acquireTimeoutMillis = -1;
// 连接池容量为8
maxConnect = 8;
// axPendingAcquires容量为Integer.MAX_VALUE
maxPendingAcquires = Integer.MAX_VALUE
releaseHealthCheck=true
//...省略无关属性
// 【重要，问题代码】
static ChannelPool fixpool =
new FixedChannelPool(b, handler, healthCheck, acquireTimeoutAction,
acquireTimeoutMillis, maxConnect, maxPendingAcquires, releaseHealthCheck, lastRecentUsed); // 【0】
// 获取连接
public Channel acquire(int timeoutMillis) throws Exception {
try {
Future<Channel> fch = fixpool.acquire(); // 【1】
// 【重要，问题代码】
Channel ch = fch.get(timeoutMillis, TimeUnit.MILLISECONDS);// 【2】
return ch;
} catch (Exception e) {
logger.error("Exception accurred when acquire channel from channel pool.", e);//【3】
throw e; //【4】
}
}
// 释放连接
public void release(Channel channel) {
try {
if (channel != null) {
fixpool.release(channel);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
分析到这里，导致“幽灵”Bug出现的原因就是获取连接任务超时后，此时还有一个异步线程在执行着从连接池获取连接的操作，这个连接取出后由于不能再正常返回给业务线程了，因为此时业务线程因为获取连接超时异常了；又因为正常情况下，释放连接的操作由业务线程来触发完成。当获取连接超时的任务从连接池取完所有可用连接后，此时服务就不可用了。而偏偏此时我们没有实现待获取连接的超时任务策略AcquireTimeoutAction,因为我们在构造一个FixedChannelPool连接池时执行的构造函数代码static ChannelPool fixpool = new FixedChannelPool(b, handler, healthCheck, acquireTimeoutAction, acquireTimeoutMillis, maxConnect, maxPendingAcquires, releaseHealthCheck, lastRecentUsed);中 传的cquireTimeoutAction参数为null且acquireTimeoutMillis为-1。也就意味着在获取连接任务超时后，没有一个定时任务会从pendingAcquireQueue队列中取出超时的获取连接任务，然后返回给业务线程！且在连接池资源耗尽的情况下，随着请求的积压，pendingAcquireQueue队列的待获取连接任务会越积越多，当超挤压的任务超出pendingAcquireQueue队列的容量后，此时就会报Too many outstanding acquire operations异常，这里pendingAcquireQueue队列容量是Integer.MAX_VALUE，因此存在OOM的风险，不过风险应该很小。
上面分析了业务线程执行Channel ch = fch.get(timeoutMillis, TimeUnit.MILLISECONDS);这句代码会因为超时而提前返回，因此即使我们实现了待获取连接超时任务策略acquireTimeoutAction也不行的，如果fch.get的超时时间timeoutMillis小于带获取连接的任务超时时间acquireTimeoutMillis同样也会导致处理待获取连接超时任务的定时任务最终获取连接后返回不了给业务线程，此时同样解决不了这个“幽灵”Bug。
5 修复获取连接超时异常导致连接池资源耗尽的Bug
相信经过前面的分析，那么如何修复这个获取连接超时异常导致连接池资源耗尽的Bug呢？相信你心中已经有了答案，下面直接上修复后的代码：
// CustomChannelPool.java
public class CustomChannelPool {
private ChannelHealthChecker healthCheck = ChannelHealthChecker.ACTIVE;
// 【修复】新建策略：当检测到获取连接超时时，此时新建一个连接
acquireTimeoutAction = AcquireTimeoutAction.NEW;
// 【修复】超时时间设置为10秒
acquireTimeoutMillis = -1;
// 【修复】连接池容量调整为100，原来的8未免太小
maxConnect = 100
// 【修复】mxPendingAcquires容量由原来的Integer.MAX_VALUE调整为100000，避免oom风险
maxPendingAcquires = 100000
releaseHealthCheck=true
//...省略无关属性
static ChannelPool fixpool =
new FixedChannelPool(b, handler, healthCheck, acquireTimeoutAction,
acquireTimeoutMillis, maxConnect, maxPendingAcquires, releaseHealthCheck, lastRecentUsed); // 【0】
// 获取连接
public Channel acquire(int timeoutMillis) throws Exception {
try {
Future<Channel> fch = fixpool.acquire(); // 【1】
// 这里连续n个获取连接超时后，因为没有归还连接，会造成连接池可用连接耗尽，最终导致服务不可用。注：n为连接池的数量
// 解决方案：需要实现AcquireTimeoutAction的NEW或FAIL策略，因为您实现的代码创建FixedChannelPool连接池时AcquireTimeoutAction参数传的是null
// Channel ch = fch.get(timeoutMillis, TimeUnit.MILLISECONDS);
// 【修复】因为需要用到AcquireTimeoutAction策略，因此这里不需要超时了
Channel ch = fch.get();
return ch;
} catch (Exception e) {
logger.error("Exception accurred when acquire channel from channel pool.", e);//【3】
throw e; //【4】
}
}
// 释放连接
public void release(Channel channel) {
try {
if (channel != null) {
fixpool.release(channel);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法