系列
Netty源码分析 - Bootstrap服务端
Netty源码分析 - Bootstrap客户端
netty源码分析 - ChannelHandler
netty源码分析 - EventLoop类关系
netty源码分析 - register分析
Netty源码分析 - NioEventLoop事件处理
netty源码分析 - accept过程分析
Netty源码分析 - ByteBuf
Netty源码分析 - 粘包和拆包问题
开篇
- 这篇文章是分析Netty Server在处理Client端连接过程中如何实现bossGroup的接受请求到workerGroup的注册Channel的过程。
- 在这篇文章中会了解bossGroup如何处理来自client端的连接，并将该连接注册到workerGroup当中。
- 在Netty源码分析 - NioEventLoop事件处理文章中分析了Netty的消息处理机制，连接请求就是消息处理的一个小分支，所以如果想了解完整的消息处理过程建议先阅读NioEventLoop事件处理这篇文章。
NioEventLoop
accept调用链
-
NioEventLoop启动消息处理线程的调用链如下：
SingleThreadEventExecutor#doStartThread => executor#execute => NioEventLoop#run -
NioEventLoop处理消息的调用链如下：
NioEventLoop#processSelectedKeys => NioEventLoop#processSelectedKeysOptimized => NioEventLoop#processSelectedKey => AbstractNioMessageChannel.NioMessageUnsafe#read -
下面的分析基于源码层面分析NioEventLoop如何处理accept事件类型。
NioEventLoop#processSelectedKey
public final class NioEventLoop extends SingleThreadEventLoop {
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
// 省略代码
try {
int readyOps = k.readyOps();
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
ch.unsafe().forceFlush();
}
// 执行读取读取事件和连接事件
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
// AbstractNioMessageChannel#NioMessageUnsafe
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
}
- processSelectedKey内部会各类事件消息，包括OP_ACCEPT事件，就是我们关心的server侧的连接请求消息类型。
- 针对OP_ACCEPT类型事件消息，由unsafe.read()继续处理，这里的unsafe指的是AbstractNioMessageChannel#NioMessageUnsafe。
AbstractNioMessageChannel#NioMessageUnsafe
public abstract class AbstractNioMessageChannel extends AbstractNioChannel {
private final class NioMessageUnsafe extends AbstractNioUnsafe {
private final List<Object> readBuf = new ArrayList<Object>();
@Override
public void read() {
assert eventLoop().inEventLoop();
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
allocHandle.reset(config);
boolean closed = false;
Throwable exception = null;
try {
try {
do {
// 读取连接请求并生成NioSocketChannel放到readBuf当中
int localRead = doReadMessages(readBuf);
// 省略代码
allocHandle.incMessagesRead(localRead);
} while (allocHandle.continueReading());
} catch (Throwable t) {
exception = t;
}
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
readPending = false;
// 触发NioServerSocketChannel的pipeline处理链
pipeline.fireChannelRead(readBuf.get(i));
}
readBuf.clear();
allocHandle.readComplete();
pipeline.fireChannelReadComplete();
// 省略代码
} finally {
// 省略代码
}
}
}
}
public class NioServerSocketChannel extends AbstractNioMessageChannel
implements io.netty.channel.socket.ServerSocketChannel {
protected int doReadMessages(List<Object> buf) throws Exception {
// 负责accept新的SocketChannel对象
SocketChannel ch = SocketUtils.accept(javaChannel());
try {
if (ch != null) {
// 封装新的SocketChannel对象为NioSocketChannel并添加到buf当中
buf.add(new NioSocketChannel(this, ch));
return 1;
}
} catch (Throwable t) {
}
return 0;
}
}
- NioMessageUnsafe#read的doReadMessages方法内部会获取来连接的SocketChannel对象，然后封装成NioSocketChannel对象放置到readBuf当中。
- 针对readBuf中的NioSocketChannel对象，逐个通过NioServerSocketChannel对应的pipeline进行处理，注意、注意、注意，是监听的NioServerSocketChannel的pipeline，相当于accept的NioSocketChannel首先是需要通过NioServerSocketChannel的pipeline的处理链进行处理的。
- 在netty源码分析 - ChannelHandler文章中已经分析过NioServerSocketChannel对应的pipeline，这里我们直接进入pipeline当中的ServerBootstrapAcceptor的handler的分析。
ServerBootstrapAcceptor
public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {
void init(Channel channel) throws Exception {
final Map<ChannelOption<?>, Object> options = options0();
synchronized (options) {
setChannelOptions(channel, options, logger);
}
final Map<AttributeKey<?>, Object> attrs = attrs0();
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();
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()));
}
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
}
- ServerBootstrap#init过程中会在NioServerSocketChannel的pipeline当中增加ServerBootstrapAcceptor对象。
- ServerBootstrapAcceptor负责处理server端的accept请求。
ServerBootstrapAcceptor#channelRead
public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {
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;
// accept的socketChannel负责绑定childHandler对象
child.pipeline().addLast(childHandler);
// accept的socketChannel负责添加具体的属性
setChannelOptions(child, childOptions, logger);
for (Entry<AttributeKey<?>, Object> e: childAttrs) {
child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
}
try {
// 注册到childGroup当中
childGroup.register(child).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
forceClose(child, future.cause());
}
}
});
} catch (Throwable t) {
forceClose(child, t);
}
}
}
}
- ServerBootstrapAcceptor#channelRead当中首先会针对新连接的NioSocketChannel的pipeline添加对应的childHandler，这个childHandler就是在初始化ServerBootstrap中指定的。
- NioSocketChannel绑定额外的属性，这些属性在初始化ServerBootstrap中指定的。
- 绑定NioSocketChannel到childGroup当中，通过childGroup.register(child)来完成，这里的childGroup在初始化ServerBootstrap中指定的。
- childGroup.register(child)实现了从bossGroup到workerGroup的注册转换。
网友评论