一、首先来看一段服务端的示例代码：
public class NettyTestServer {
public void bind(int port) throws Exception{
EventLoopGroup bossgroup = new NioEventLoopGroup();//创建BOSS线程组
EventLoopGroup workgroup = new NioEventLoopGroup();//创建WORK线程组
try{
ServerBootstrap b = new ServerBootstrap();
b.group(bossgroup,workgroup)//绑定BOSS和WORK线程组
.channel(NioServerSocketChannel.class)//设置channel类型，服务端用的是NioServerSocketChannel
.option(ChannelOption.SO_BACKLOG,100) //设置channel的配置选项
.handler(new LoggingHandler(LogLevel.INFO))//设置NioServerSocketChannel的Handler
.childHandler(new ChannelInitializer<SocketChannel>() {//设置childHandler，作为新建的NioSocketChannel的初始化Handler
@Override//当新建的与客户端通信的NioSocketChannel被注册到EventLoop成功时，该方法会被调用，用于添加业务Handler
protected void initChannel(SocketChannel ch) throws Exception {
ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new EchoServerHandler());
}
});
ChannelFuture f = b.bind(port).sync();//同步等待绑定结束
f.channel().closeFuture().sync();//同步等待关闭
}finally {
bossgroup.shutdownGracefully();
workgroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception{
int port = 8082;
new NettyTestServer().bind(port);
}
}
@ChannelHandler.Sharable
class EchoServerHandler extends ChannelInboundHandlerAdapter{
int count = 0;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String body = (String)msg;
System.out.println("This is" + ++count + "times receive client:[" + body + "]");
body += "$_";
ByteBuf echo = Unpooled.copiedBuffer(body.getBytes());
ctx.writeAndFlush(echo);
ctx.fireChannelRead("my name is chenyang");
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
二、首先来看一下ServerBootstrap类，顾名思义，它是一个服务端启动类，用于帮助用户快速配置、启动服务端服务。先来看一下该类的主要成员定义：
/**
* {@link Bootstrap} sub-class which allows easy bootstrap of {@link ServerChannel}
*
*/
public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> { private static final InternalLogger logger = InternalLoggerFactory.getInstance(ServerBootstrap.class);
//以下都是针对NioSocketChannel的
private final Map<ChannelOption<?>, Object> childOptions = new LinkedHashMap<ChannelOption<?>, Object>();
private final Map<AttributeKey<?>, Object> childAttrs = new LinkedHashMap<AttributeKey<?>, Object>();
private volatile EventLoopGroup childGroup;
private volatile ChannelHandler childHandler;
可见，ServerBootstrap是AbstractBootstrap的子类，AbstractBootstrap的成员主要有：
/**
* {@link AbstractBootstrap} is a helper class that makes it easy to bootstrap a {@link Channel}. It support
* method-chaining to provide an easy way to configure the {@link AbstractBootstrap}.
*
* <p>When not used in a {@link ServerBootstrap} context, the {@link #bind()} methods are useful for connectionless
* transports such as datagram (UDP).</p>
*/
public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {
//以下都是针对服务端NioServerSocketChannel的
volatile EventLoopGroup group;
private volatile ChannelFactory<? extends C> channelFactory;
private volatile SocketAddress localAddress;
private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();
private final Map<AttributeKey<?>, Object> attrs = new LinkedHashMap<AttributeKey<?>, Object>();
private volatile ChannelHandler handler;
用一张图说明两个类之间的关系如下（原图出自：http://blog.csdn.net/zxhoo/article/details/17532857）。
总结如下： ServerBootstrap比AbstractBootstrap多了4个Part，其中AbstractBootstrap的成员用于设置服务端NioServerSocketChannel（包括所使用的线程组、使用的channel工厂类、使用的Handler以及地址和选项信息等）， ServerBootstrap的4个成员用于设置为有新连接时新建的NioSocketChannel。
三、ServerBootstrap配置源码解释
1）b.group(bossgroup,workgroup)
/**
* Set the {@link EventLoopGroup} for the parent (acceptor) and the child (client). These
* {@link EventLoopGroup}'s are used to handle all the events and IO for {@link ServerChannel} and
* {@link Channel}'s.
*/
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
super.group(parentGroup);//设置BOSS线程组（在AbstractBootstrap中）
if (childGroup == null) {
throw new NullPointerException("childGroup");
}
if (this.childGroup != null) {
throw new IllegalStateException("childGroup set already");
}
this.childGroup = childGroup;//设置WORK线程组
return this;
}
2） .channel(NioServerSocketChannel.class)
/**
* The {@link Class} which is used to create {@link Channel} instances from.
* You either use this or {@link #channelFactory(ChannelFactory)} if your
* {@link Channel} implementation has no no-args constructor.
*/
public B channel(Class<? extends C> channelClass) {
if (channelClass == null) {
throw new NullPointerException("channelClass");
}
return channelFactory(new BootstrapChannelFactory<C>(channelClass));//设置channel工厂
}
channelFactory方法就是用来设置channel工厂的，这里的工厂就是BootstrapChannelFactory（是一个泛型类）。
/**
* {@link ChannelFactory} which is used to create {@link Channel} instances from
* when calling {@link #bind()}. This method is usually only used if {@link #channel(Class)}
* is not working for you because of some more complex needs. If your {@link Channel} implementation
* has a no-args constructor, its highly recommend to just use {@link #channel(Class)} for
* simplify your code.
*/
@SuppressWarnings("unchecked")
public B channelFactory(ChannelFactory<? extends C> channelFactory) {
if (channelFactory == null) {
throw new NullPointerException("channelFactory");
}
if (this.channelFactory != null) {
throw new IllegalStateException("channelFactory set already");
} this.channelFactory = channelFactory;//设置channel工厂
return (B) this;
}
下面就是channel工厂类的实现，构造函数传入一个channel类型（针对服务端也就是NioServerSocketChannel.class）,BootstrapChannelFactory工厂类提供的newChannel方法将使用反射创建对应的channel。用于channel的创建一般只在启动的时候进行，因此使用反射不会造成性能的问题。
private static final class BootstrapChannelFactory<T extends Channel> implements ChannelFactory<T> {
private final Class<? extends T> clazz;
BootstrapChannelFactory(Class<? extends T> clazz) {
this.clazz = clazz;
}
@Override
public T newChannel() {//需要创建channel的时候，次方法将被调用
try {
return clazz.newInstance();//反射创建对应channel
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + clazz, t);
}
}
@Override
public String toString() {
return StringUtil.simpleClassName(clazz) + ".class";
}
}
3） .option(ChannelOption.SO_BACKLOG,100)
用来设置channel的选项，比如设置BackLog的大小等。
/**
* Allow to specify a {@link ChannelOption} which is used for the {@link Channel} instances once they got
* created. Use a value of {@code null} to remove a previous set {@link ChannelOption}.
*/
@SuppressWarnings("unchecked")
public <T> B option(ChannelOption<T> option, T value) {
if (option == null) {
throw new NullPointerException("option");
}
if (value == null) {
synchronized (options) {
options.remove(option);
}
} else {
synchronized (options) {
options.put(option, value);
}
}
return (B) this;
}
4） .handler(new LoggingHandler(LogLevel.INFO))
用于设置服务端NioServerSocketChannel的Handler。
/**
* the {@link ChannelHandler} to use for serving the requests.
*/
@SuppressWarnings("unchecked")
public B handler(ChannelHandler handler) {
if (handler == null) {
throw new NullPointerException("handler");
}
this.handler = handler;//设置的是父类AbstractBootstrap里的成员，也就是该handler是被NioServerSocketChannel使用
return (B) this;
}
5） .childHandler(new ChannelInitializer<SocketChannel>() {
一定要分清.handler和.childHandler的区别，首先，两者都是设置一个Handler，但是，前者设置的Handler是属于服务端NioServerSocketChannel的，而后者设置的Handler是属于每一个新建的NioSocketChannel的（每当有一个来自客户端的连接时，否会创建一个新的NioSocketChannel）。
/**
* Set the {@link ChannelHandler} which is used to serve the request for the {@link Channel}'s.
*/
public ServerBootstrap childHandler(ChannelHandler childHandler) {
if (childHandler == null) {
throw new NullPointerException("childHandler");
}
this.childHandler = childHandler;
return this;
}
至此，ServerBootstrap的配置完成，其实有人可能会很好奇，为什么不直接在ServerBootstrap的构造函数中一步完成这些初始化配置操作，这样做虽然可以，但是这会导致ServerBootstrap构造函数的参数过多，而是用Builder模式（也就是ServerBootstrap目前采用的模式，可以参见<<effective java>>）则可以有效的解决构造方法参数过多的问题。
四、bind流程
1）一切从bind开始 ChannelFuture f = b.bind(port).sync();
/**
* Create a new {@link Channel} and bind it.
*/
public ChannelFuture bind(int inetPort) {
return bind(new InetSocketAddress(inetPort));
}
继续深入bind
/**
* Create a new {@link Channel} and bind it.
*/
public ChannelFuture bind(SocketAddress localAddress) {
validate();
if (localAddress == null) {
throw new NullPointerException("localAddress");
}
return doBind(localAddress);
}
继续摄入doBind
private ChannelFuture doBind(final SocketAddress localAddress) {
final ChannelFuture regFuture = initAndRegister();//初始化并注册一个channel
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
//等待注册成功
if (regFuture.isDone()) {
// At this point we know that the registration was complete and successful.
ChannelPromise promise = channel.newPromise();
doBind0(regFuture, channel, localAddress, promise);//执行channel.bind()
return promise;
} else {
// Registration future is almost always fulfilled already, but just in case it's not.
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Throwable cause = future.cause();
if (cause != null) {
// Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
// IllegalStateException once we try to access the EventLoop of the Channel.
promise.setFailure(cause);
} else {
// Registration was successful, so set the correct executor to use.
// See https://github.com/netty/netty/issues/2586
promise.executor = channel.eventLoop();
}
doBind0(regFuture, channel, localAddress, promise);
}
});
return promise;
}
}
doBind中最重要的一步就是调用initAndRegister方法了，它会初始化并注册一个channel，直接看源码吧。
final ChannelFuture initAndRegister() {
final Channel channel = channelFactory().newChannel();//还记得前面我们设置过channel工厂么，终于排上用场了
try {
init(channel);//初始化channel（就是NioServerSocketChannel）
} catch (Throwable t) {
channel.unsafe().closeForcibly();
// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
}
ChannelFuture regFuture = group().register(channel);//向EventLoopGroup中注册一个channel
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
// If we are here and the promise is not failed, it's one of the following cases:
// 1) If we attempted registration from the event loop, the registration has been completed at this point.
// i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
// 2) If we attempted registration from the other thread, the registration request has been successfully
// added to the event loop's task queue for later execution.
// i.e. It's safe to attempt bind() or connect() now:
// because bind() or connect() will be executed *after* the scheduled registration task is executed
// because register(), bind(), and connect() are all bound to the same thread.
return regFuture;
}
先来看一下init方法
@Override
void init(Channel channel) throws Exception {
final Map<ChannelOption<?>, Object> options = options();
synchronized (options) {
channel.config().setOptions(options);//设置之前配置的channel选项
} final Map<AttributeKey<?>, Object> attrs = attrs();
synchronized (attrs) {
for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
@SuppressWarnings("unchecked")
AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
channel.attr(key).set(e.getValue());//设置之前配置的属性
}
} ChannelPipeline p = channel.pipeline();//获取channel绑定的pipeline（pipeline实在channel创建的时候创建并绑定的）
if (handler() != null) {//如果用户配置过Handler
p.addLast(handler());//为NioServerSocketChannel绑定的pipeline添加Handler
}
//开始准备child用到的4个part，因为接下来就要使用它们。
final EventLoopGroup currentChildGroup = childGroup;
final ChannelHandler currentChildHandler = childHandler;
final Entry<ChannelOption<?>, Object>[] currentChildOptions;
final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
synchronized (childOptions) {
currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
}
synchronized (childAttrs) {
currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
}
//为NioServerSocketChannel的pipeline添加一个初始化Handler,当NioServerSocketChannel在EventLoop注册成功时，该handler的init方法将被调用
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(new ServerBootstrapAcceptor(//为NioServerSocketChannel的pipeline添加ServerBootstrapAcceptor处理器
//该Handler主要用来将新创建的NioSocketChannel注册到EventLoopGroup中
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
init执行之后，接下来看一下注册过程（ChannelFuture regFuture = group().register(channel); 注意，这里的group是之前设置的BOSS EventLoopGroup）
@Override
public ChannelFuture register(Channel channel) {
return next().register(channel);//首先使用next()在BOSS EventLoopGroup中选出下一个EventLoop，然后执行注册
}
@Override
public ChannelFuture register(Channel channel) {
return register(channel, new DefaultChannelPromise(channel, this));
}
@Override
public ChannelFuture register(final Channel channel, final ChannelPromise promise) {
if (channel == null) {
throw new NullPointerException("channel");
}
if (promise == null) {
throw new NullPointerException("promise");
} channel.unsafe().register(this, promise);//unsafe执行的都是实际的操作
return promise;
}
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}
if (!isCompatible(eventLoop)) {
promise.setFailure(
new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
return;
} AbstractChannel.this.eventLoop = eventLoop;//绑定为该channel选的的EventLoop
//必须保证注册是由该EventLoop发起的，否则会单独封装成一个Task，由该EventLoop执行
if (eventLoop.inEventLoop()) {
register0(promise);//注册
} else {
try {
eventLoop.execute(new OneTimeTask() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
logger.warn(
"Force-closing a channel whose registration task was not accepted by an event loop: {}",
AbstractChannel.this, t);
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
}
private void register0(ChannelPromise promise) {
try {
// check if the channel is still open as it could be closed in the mean time when the register
// call was outside of the eventLoop
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
boolean firstRegistration = neverRegistered;
doRegister();//最底层的注册调用
neverRegistered = false;
registered = true;
safeSetSuccess(promise);//设置注册结果为成功
pipeline.fireChannelRegistered();//发起pipeline调用fireChannelRegistered（head.fireChannelRegistered）
// Only fire a channelActive if the channel has never been registered. This prevents firing
// multiple channel actives if the channel is deregistered and re-registered.
if (firstRegistration && isActive()) {//如果是首次注册，而且channel已经处于Active状态（如果是服务端，表示listen成功，如果是客户端，便是connect成功）
pipeline.fireChannelActive();//发起pipeline的fireChannelActive
}
} catch (Throwable t) {
// Close the channel directly to avoid FD leak.
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
doRegister会完成在EventLoop的Selector上的注册任务。
@Override
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
selectionKey = javaChannel().register(eventLoop().selector, 0, this);//注意，此时op位为0，channel还不能监听读写事件
return;
} catch (CancelledKeyException e) {
if (!selected) {
// Force the Selector to select now as the "canceled" SelectionKey may still be
// cached and not removed because no Select.select(..) operation was called yet.
eventLoop().selectNow();
selected = true;
} else {
// We forced a select operation on the selector before but the SelectionKey is still cached
// for whatever reason. JDK bug ?
throw e;
}
}
}
}
由上可知，注册成功后，NioServerSocketChannel还不能监听读写事件，那么什么时候回开始监听呢？由于注册成功之后，会进行pipeline.fireChannelRegistered()调用，该事件会在NioServerSocketChannel的pipeline中传播（从head开始，逐步findContextInbound），这会导致Inbound类型的Handler的channelRegistered方法被调用。还记得在init方法中为NioServerSocketChannel添加的ChannelInitializer的Handler吗，它也是一个InboundHandler，看一下他的实现：
@Sharable
public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter { private static final InternalLogger logger = InternalLoggerFactory.getInstance(ChannelInitializer.class); /**
* This method will be called once the {@link Channel} was registered. After the method returns this instance
* will be removed from the {@link ChannelPipeline} of the {@link Channel}.
*
* @param ch the {@link Channel} which was registered.
* @throws Exception is thrown if an error occurs. In that case the {@link Channel} will be closed.
*/
protected abstract void initChannel(C ch) throws Exception;//抽象方法，由子类实现 @Override
@SuppressWarnings("unchecked")
public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {//该方法会在NioServerScoketChannel注册成功时被调用
ChannelPipeline pipeline = ctx.pipeline();
boolean success = false;
try {
initChannel((C) ctx.channel());//调用initChannel
pipeline.remove(this);//初始化Handler只完成初始化工作，初始化完成自后就把自己删除
ctx.fireChannelRegistered();//继续传播channelRegistered事件
success = true;
} catch (Throwable t) {
logger.warn("Failed to initialize a channel. Closing: " + ctx.channel(), t);
} finally {
if (pipeline.context(this) != null) {
pipeline.remove(this);
}
if (!success) {
ctx.close();
}
}
}
}
在重复贴一次代码，看一下initChannel里面是什么
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(Channel ch) throws Exception {//被channelRegistered调用
ch.pipeline().addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
}
可以看到，initChannel只是向pipeline中添加了ServerBootstrapAcceptor类型的Handler。
但是这还是没有看到给NioServerSocketChannel注册读写事件的地方，继续看之前的register0代码，它还会调用pipleline的fireChannelActive方法，看一下该方方法的代码：
@Override
public ChannelPipeline fireChannelActive() {
head.fireChannelActive();//将ChannelActive事件在pipeline中传播
//如果channel被配置成自动可读的，那么久发起读事件
if (channel.config().isAutoRead()) {
channel.read();//pipeline.read()-->tail.read()-->*****-->head.read()-->unsafe.beginRead()
} return this;
}
@Override
public ChannelHandlerContext fireChannelActive() {//head的fireChannelActive()
final AbstractChannelHandlerContext next = findContextInbound();//寻找下一个Inbound类型的Context
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
next.invokeChannelActive();//调用Context中的Handler的channelActive方法
} else {
executor.execute(new OneTimeTask() {
@Override
public void run() {
next.invokeChannelActive();
}
});
}
return this;
}
看一下beginRead实现：
@Override
public final void beginRead() {
if (!isActive()) {
return;
} try {
doBeginRead();//真正的注册读事件
} catch (final Exception e) {
invokeLater(new OneTimeTask() {
@Override
public void run() {
pipeline.fireExceptionCaught(e);
}
});
close(voidPromise());
}
}
@Override
protected void doBeginRead() throws Exception {
// Channel.read() or ChannelHandlerContext.read() was called
if (inputShutdown) {
return;
} final SelectionKey selectionKey = this.selectionKey;
if (!selectionKey.isValid()) {
return;
} readPending = true; final int interestOps = selectionKey.interestOps();
if ((interestOps & readInterestOp) == 0) {
selectionKey.interestOps(interestOps | readInterestOp);//真正的注册读事件
}
}
五、客户端接入过程
接下来看看，当一个客户端连接进来时，都发生了什么。
1）首先从事件的源头看起，下面是EventLoop的事件循环
@Override
protected void run() {
for (;;) {
boolean oldWakenUp = wakenUp.getAndSet(false);
try {
if (hasTasks()) {
selectNow();
} else {
select(oldWakenUp);//调用selector.select() // 'wakenUp.compareAndSet(false, true)' is always evaluated
// before calling 'selector.wakeup()' to reduce the wake-up
// overhead. (Selector.wakeup() is an expensive operation.)
//
// However, there is a race condition in this approach.
// The race condition is triggered when 'wakenUp' is set to
// true too early.
//
// 'wakenUp' is set to true too early if:
// 1) Selector is waken up between 'wakenUp.set(false)' and
// 'selector.select(...)'. (BAD)
// 2) Selector is waken up between 'selector.select(...)' and
// 'if (wakenUp.get()) { ... }'. (OK)
//
// In the first case, 'wakenUp' is set to true and the
// following 'selector.select(...)' will wake up immediately.
// Until 'wakenUp' is set to false again in the next round,
// 'wakenUp.compareAndSet(false, true)' will fail, and therefore
// any attempt to wake up the Selector will fail, too, causing
// the following 'selector.select(...)' call to block
// unnecessarily.
//
// To fix this problem, we wake up the selector again if wakenUp
// is true immediately after selector.select(...).
// It is inefficient in that it wakes up the selector for both
// the first case (BAD - wake-up required) and the second case
// (OK - no wake-up required). if (wakenUp.get()) {
selector.wakeup();
}
} cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
processSelectedKeys();
runAllTasks();
} else {
final long ioStartTime = System.nanoTime(); processSelectedKeys();//有事件发生时，执行这里 final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
} if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
break;
}
}
} catch (Throwable t) {
logger.warn("Unexpected exception in the selector loop.", t); // Prevent possible consecutive immediate failures that lead to
// excessive CPU consumption.
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// Ignore.
}
}
}
}
看一下processSelectedKeys代码
private void processSelectedKeys() {
if (selectedKeys != null) {
processSelectedKeysOptimized(selectedKeys.flip());//执行这里
} else {
processSelectedKeysPlain(selector.selectedKeys());
}
}
private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
for (int i = 0;; i ++) {
final SelectionKey k = selectedKeys[i];
if (k == null) {
break;
}
// null out entry in the array to allow to have it GC'ed once the Channel close
// See https://github.com/netty/netty/issues/2363
selectedKeys[i] = null;
final Object a = k.attachment();
if (a instanceof AbstractNioChannel) {//因为是NioServerSocketChannel,所以执行这里
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
if (needsToSelectAgain) {
// null out entries in the array to allow to have it GC'ed once the Channel close
// See https://github.com/netty/netty/issues/2363
for (;;) {
if (selectedKeys[i] == null) {
break;
}
selectedKeys[i] = null;
i++;
}
selectAgain();
// Need to flip the optimized selectedKeys to get the right reference to the array
// and reset the index to -1 which will then set to 0 on the for loop
// to start over again.
//
// See https://github.com/netty/netty/issues/1523
selectedKeys = this.selectedKeys.flip();
i = -1;
}
}
}
private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
// close the channel if the key is not valid anymore
unsafe.close(unsafe.voidPromise());
return;
}
try {
int readyOps = k.readyOps();
// Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
// to a spin loop
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();//因为是ACCEPT事件，所以执行这里（这里的read会因为NioServerSocketChannel和NioSocketChannel不同）
if (!ch.isOpen()) {
// Connection already closed - no need to handle write.
return;
}
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
ch.unsafe().forceFlush();
}
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
// See https://github.com/netty/netty/issues/924
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
NioServerSocketChannel继承了AbstractNioMessageChannel，所以执行的是AbstractNioMessageChannel的版本
@Override
public void read() {
assert eventLoop().inEventLoop();
final ChannelConfig config = config();
if (!config.isAutoRead() && !isReadPending()) {
// ChannelConfig.setAutoRead(false) was called in the meantime
removeReadOp();
return;
} final int maxMessagesPerRead = config.getMaxMessagesPerRead();
final ChannelPipeline pipeline = pipeline();//获取服务端NioServerSocketChannel的pipeline
boolean closed = false;
Throwable exception = null;
try {
try {
for (;;) {
int localRead = doReadMessages(readBuf);//执行这里
if (localRead == 0) {
break;
}
if (localRead < 0) {
closed = true;
break;
} // stop reading and remove op
if (!config.isAutoRead()) {
break;
} if (readBuf.size() >= maxMessagesPerRead) {
break;
}
}
} catch (Throwable t) {
exception = t;
}
setReadPending(false);
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
pipeline.fireChannelRead(readBuf.get(i));//引发ChannelRead
} readBuf.clear();
pipeline.fireChannelReadComplete();//引发channelReadComplete if (exception != null) {
if (exception instanceof IOException && !(exception instanceof PortUnreachableException)) {
// ServerChannel should not be closed even on IOException because it can often continue
// accepting incoming connections. (e.g. too many open files)
closed = !(AbstractNioMessageChannel.this instanceof ServerChannel);
} pipeline.fireExceptionCaught(exception);
} if (closed) {
if (isOpen()) {
close(voidPromise());
}
}
} 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 (!config.isAutoRead() && !isReadPending()) {
removeReadOp();
}
}
}
而对于NioSocketChannel而言，其继承自AbstractNioByteChannel,因此调用的AbstractNioByteChannel的read版本如下：
@Override
public final void read() {
final ChannelConfig config = config();
if (!config.isAutoRead() && !isReadPending()) {
// ChannelConfig.setAutoRead(false) was called in the meantime
removeReadOp();
return;
} final ChannelPipeline pipeline = pipeline();
final ByteBufAllocator allocator = config.getAllocator();
final int maxMessagesPerRead = config.getMaxMessagesPerRead();
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
} ByteBuf byteBuf = null;
int messages = 0;
boolean close = false;
try {
int totalReadAmount = 0;//读到的总长度
boolean readPendingReset = false;
do {
byteBuf = allocHandle.allocate(allocator);
int writable = byteBuf.writableBytes();//获取bytebuf还可以写入的字节数
int localReadAmount = doReadBytes(byteBuf);//真正的读取，localReadAmount本次读取的实际长度
if (localReadAmount <= 0) {//什么都没有读到
// not was read release the buffer
byteBuf.release();
byteBuf = null;
close = localReadAmount < 0;
break;//跳出循环
}
if (!readPendingReset) {
readPendingReset = true;
setReadPending(false);
}
pipeline.fireChannelRead(byteBuf);//发起调用channelRead，将bytebuf传过去
byteBuf = null;
//如果当前读到的总长度+本次读到的总长度已经大于Integer类型的最大值
if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) {
// Avoid overflow.
totalReadAmount = Integer.MAX_VALUE;
break;//跳出循环
}
//更新总长度
totalReadAmount += localReadAmount; // stop reading
if (!config.isAutoRead()) {
break;//如果不是自动读取，那么读取一次之后就自动停止了
}
//如果本次读取的大小没有把bytebuf填满，那么说明数据已经全部读取了
if (localReadAmount < writable) {
// Read less than what the buffer can hold,
// which might mean we drained the recv buffer completely.
break;//跳出循环
}
} while (++ messages < maxMessagesPerRead); pipeline.fireChannelReadComplete();//跳出循环后，引发channelReadComplete
allocHandle.record(totalReadAmount); if (close) {
closeOnRead(pipeline);
close = false;
}
} catch (Throwable t) {
handleReadException(pipeline, byteBuf, t, close);
} 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 (!config.isAutoRead() && !isReadPending()) {
removeReadOp();
}
}
}
接着看doMessages
@Override
protected int doReadMessages(List<Object> buf) throws Exception {
SocketChannel ch = javaChannel().accept();//创建SocketChannel,accept客户端 try {
if (ch != null) {
buf.add(new NioSocketChannel(this, ch));
return 1;
}
} catch (Throwable t) {
logger.warn("Failed to create a new channel from an accepted socket.", t); try {
ch.close();
} catch (Throwable t2) {
logger.warn("Failed to close a socket.", t2);
}
} return 0;
}
执行完doReadMessages之后，针对客户端的SocketChannel已经创建了，由于之后还会引发channelRead和channelReadComplete事件，而这些都会导致pipeline中的ServerBootstrapAcceptor的相应方法被调用，来看一下ServerBootstrapAcceptor源码：
private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {
private final EventLoopGroup childGroup;
private final ChannelHandler childHandler;
private final Entry<ChannelOption<?>, Object>[] childOptions;
private final Entry<AttributeKey<?>, Object>[] childAttrs;
ServerBootstrapAcceptor(
EventLoopGroup childGroup, ChannelHandler childHandler,
Entry<ChannelOption<?>, Object>[] childOptions, Entry<AttributeKey<?>, Object>[] childAttrs) {
this.childGroup = childGroup;
this.childHandler = childHandler;
this.childOptions = childOptions;
this.childAttrs = childAttrs;
}
@Override
@SuppressWarnings("unchecked")
public void channelRead(ChannelHandlerContext ctx, Object msg) {
final Channel child = (Channel) msg;
child.pipeline().addLast(childHandler);//将最开始配置的childHandler添加到SocketChannel的pipeline中，这个Handler也是一个初始化Handler，原理和服务端的一致
for (Entry<ChannelOption<?>, Object> e: childOptions) {
try {
if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {
logger.warn("Unknown channel option: " + e);
}
} catch (Throwable t) {
logger.warn("Failed to set a channel option: " + child, t);
}
}
for (Entry<AttributeKey<?>, Object> e: childAttrs) {
child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
}
try {
childGroup.register(child).addListener(new ChannelFutureListener() {//将SocketChannel注册到WORK EventLoopGroup中，注册过程与服务端类似，此处不再讲解
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
forceClose(child, future.cause());
}
}
});
} catch (Throwable t) {
forceClose(child, t);
}
}
private static void forceClose(Channel child, Throwable t) {
child.unsafe().closeForcibly();
logger.warn("Failed to register an accepted channel: " + child, t);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
final ChannelConfig config = ctx.channel().config();
if (config.isAutoRead()) {
// stop accept new connections for 1 second to allow the channel to recover
// See https://github.com/netty/netty/issues/1328
config.setAutoRead(false);
ctx.channel().eventLoop().schedule(new Runnable() {
@Override
public void run() {
config.setAutoRead(true);
}
}, 1, TimeUnit.SECONDS);
}
// still let the exceptionCaught event flow through the pipeline to give the user
// a chance to do something with it
ctx.fireExceptionCaught(cause);
}
}
引用一张图（出自：http://blog.csdn.net/zxhoo/article/details/17532857） 。