netty源码之断开连接
Posted better_hui
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了netty源码之断开连接相关的知识，希望对你有一定的参考价值。
目录
2、doClose0调用jdk的close 关闭channel
3、fireChannelInactiveAndDeregister 传播关闭、解绑定事件
4、doDeregister取消selector上的selectorKey
一、前言
关闭的逻辑是在read方法里 ， 当接受的数据<0时，说明没有可接受的数据了，可以关闭了。
二、正常关闭
1.allocHandle.lastBytesRead() <= 0这一行会发现数据小于零，释放byteBuf 2.进入到closeOnRead(pipeline);完成关闭事件
@Override
public final void read() {
final ChannelConfig config = config();
if (shouldBreakReadReady(config)) {
clearReadPending();
return;
}
final ChannelPipeline pipeline = pipeline();
// 内存分配器
final ByteBufAllocator allocator = config.getAllocator();
// 接收数据测handler
final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
allocHandle.reset(config);
ByteBuf byteBuf = null;
boolean close = false;
try {
do {
// 分配内存是自适应的
byteBuf = allocHandle.allocate(allocator);
// 开始读取数据
allocHandle.lastBytesRead(doReadBytes(byteBuf));
if (allocHandle.lastBytesRead() <= 0) {// 读取为-1，表示断开连接
// nothing was read. release the buffer.
byteBuf.release();//释放byteBuf的空间
byteBuf = null;
close = allocHandle.lastBytesRead() < 0;
if (close) {
// There is nothing left to read as we received an EOF.
readPending = false;
}
break;
}
allocHandle.incMessagesRead(1);//读取一次数据
readPending = false;
pipeline.fireChannelRead(byteBuf);// 将读到的数据传递出去
byteBuf = null;
} while (allocHandle.continueReading());//继续读取
// 通过当前
allocHandle.readComplete();//计算下一次的需要分配的空间
pipeline.fireChannelReadComplete();// 将完成读取的事件传递出去
if (close) {
closeOnRead(pipeline);//关闭
}
} catch (Throwable t) {
handleReadException(pipeline, byteBuf, t, close, allocHandle);
} finally {
// Check if there is a readPending which was not processed yet.
// This could be for two reasons:
// * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
// * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
//
// See https://github.com/netty/netty/issues/2254
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
}
继续跟进核心的方法在AbstractChannel.close方法里
private void close(final ChannelPromise promise, final Throwable cause, final ClosedChannelException closeCause, final boolean notify) { if (!promise.setUncancellable()) { return; }
if (closeInitiated) {
if (closeFuture.isDone()) {
// Closed already.
safeSetSuccess(promise);
} else if (!(promise instanceof VoidChannelPromise)) { // Only needed if no VoidChannelPromise.
// This means close() was called before so we just register a listener and return
closeFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
promise.setSuccess();
}
});
}
return;
}
closeInitiated = true;
final boolean wasActive = isActive();
final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
// outboundBuffer置空
this.outboundBuffer = null; // Disallow adding any messages and flushes to outboundBuffer.
Executor closeExecutor = prepareToClose();// 关闭之前将数据发送完成之前的一些操作
if (closeExecutor != null) {
closeExecutor.execute(new Runnable() {
@Override
public void run() {
try {
// Execute the close.
doClose0(promise);
} finally {
// Call invokeLater so closeAndDeregister is executed in the EventLoop again!
invokeLater(new Runnable() {
@Override
public void run() {
if (outboundBuffer != null) {
// Fail all the queued messages
outboundBuffer.failFlushed(cause, notify);
outboundBuffer.close(closeCause);
}
fireChannelInactiveAndDeregister(wasActive);
}
});
}
}
});
} else {
try {
// Close the channel and fail the queued messages in all cases.
doClose0(promise);// 执行close操作
} finally {
// 强行关闭outboundBuffer
if (outboundBuffer != null) {
// Fail all the queued messages.
outboundBuffer.failFlushed(cause, notify);
outboundBuffer.close(closeCause);
}
}
if (inFlush0) {
invokeLater(new Runnable() {
@Override
public void run() {
fireChannelInactiveAndDeregister(wasActive);
}
});
} else {
// 将channel不活跃的事件传播出去
fireChannelInactiveAndDeregister(wasActive);
}
}
}
1、关闭前的处理工作
1.首先判断是否需要阻塞，先将数据发送完成 2.doDeregister()使得所有的selectorKey失效，防止关闭的时候注册事件到selector上 3.返回一个GlobalEventExecutor.INSTANCE线程池执行任务
2、doClose0调用jdk的close 关闭channel
3、fireChannelInactiveAndDeregister 传播关闭、解绑定事件
1.cancel所有的selectKey 2.pipeline.fireChannelInactive()将close事件传播出去 3.pipeline也取消注册
private void deregister(final ChannelPromise promise, final boolean fireChannelInactive) {
if (!promise.setUncancellable()) {
return;
}
if (!registered) {
safeSetSuccess(promise);
return;
}
// As a user may call deregister() from within any method while doing processing in the ChannelPipeline,
// we need to ensure we do the actual deregister operation later. This is needed as for example,
// we may be in the ByteToMessageDecoder.callDecode(...) method and so still try to do processing in
// the old EventLoop while the user already registered the Channel to a new EventLoop. Without delay,
// the deregister operation this could lead to have a handler invoked by different EventLoop and so
// threads.
//
// See:
// https://github.com/netty/netty/issues/4435
invokeLater(new Runnable() {
@Override
public void run() {
try {
doDeregister();// cancel所有的selectKey
} catch (Throwable t) {
logger.warn("Unexpected exception occurred while deregistering a channel.", t);
} finally {
if (fireChannelInactive) {
// 最后将这个关闭事件传播出去
pipeline.fireChannelInactive();
}
// Some transports like local and AIO does not allow the deregistration of
// an open channel. Their doDeregister() calls close(). Consequently,
// close() calls deregister() again - no need to fire channelUnregistered, so check
// if it was registered.
if (registered) {
registered = false;
// pipeline也取消注册
pipeline.fireChannelUnregistered();
}
safeSetSuccess(promise);
}
}
});
}
4、doDeregister取消selector上的selectorKey
它本质上就是取消了selector上的所有的selectorKey，但是做出了一些优化
三、异常关闭
进入AbstractNioByteChannel，其中的read方法为例,假设正在读取数据的时候突然抛出异常，会被handleReadException(pipeline, byteBuf, t, close, allocHandle);处理，并关闭channel。
@Override
public final void read() {
final ChannelConfig config = config();
if (shouldBreakReadReady(config)) {
clearReadPending();
return;
}
final ChannelPipeline pipeline = pipeline();
// 内存分配器
final ByteBufAllocator allocator = config.getAllocator();
// 接收数据测handler
final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
allocHandle.reset(config);
ByteBuf byteBuf = null;
boolean close = false;
try {
do {
// 分配内存是自适应的
byteBuf = allocHandle.allocate(allocator);
// 开始读取数据
allocHandle.lastBytesRead(doReadBytes(byteBuf));
if (allocHandle.lastBytesRead() <= 0) {// 读取为-1，表示断开连接
// nothing was read. release the buffer.
byteBuf.release();//释放byteBuf的空间
byteBuf = null;
close = allocHandle.lastBytesRead() < 0;
if (close) {
// There is nothing left to read as we received an EOF.
readPending = false;
}
break;
}
allocHandle.incMessagesRead(1);//读取一次数据
readPending = false;
pipeline.fireChannelRead(byteBuf);// 将读到的数据传递出去
byteBuf = null;
} while (allocHandle.continueReading());//继续读取
// 通过当前
allocHandle.readComplete();//计算下一次的需要分配的空间
pipeline.fireChannelReadComplete();// 将完成读取的事件传递出去
if (close) {
closeOnRead(pipeline);//关闭
}
} catch (Throwable t) {
handleReadException(pipeline, byteBuf, t, close, allocHandle);
} finally {
// Check if there is a readPending which was not processed yet.
// This could be for two reasons:
// * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
// * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
//
// See https://github.com/netty/netty/issues/2254
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
四、总结
以上是关于netty源码之断开连接的主要内容，如果未能解决你的问题，请参考以下文章