网站域名地址查询,wordpress属于区域连技术吗,wordpress 查看原图,免费的产品推广平台背景
本文介绍Netty的通道组件NioServerSocketChannel和NioSocketChannel#xff0c;从源码的角度介绍其实现原理。
1.NioServerSocketChannel
Netty本质是对NIO的封装和增强#xff0c;因此Netty框架中必然包含了对于ServerSocketChannel的构建、配置以及向选择器注册从源码的角度介绍其实现原理。
1.NioServerSocketChannel
Netty本质是对NIO的封装和增强因此Netty框架中必然包含了对于ServerSocketChannel的构建、配置以及向选择器注册如下所示:
// 创建ServerSocketChannel对象
ServerSocketChannel serverSocketChannel SelectorProvider.provider().openServerSocketChannel();// ServerSocketChannel通道设置为非阻塞
serverSocketChannel.configureBlocking(false);// 将ServerSocketChannel通道注册至选择器
serverSocketChannel.register(Selector, opts, attachment);// 接收客户端连接得到SocketChannel通道
SocketChannel socketChannel serverSocketChannel.accept();其中的构建和配置过程发生在NioServerSocketChannel的实例化过程。
1.1 NioServerSocketChannel构造函数 NioServerSocketChannel实例化过程包含了对serverSocketChannel的创建以及配置 Netty启动时通过反射调用NioServerSocketChannel的无参构造函数创建NioServerSocketChannel对象.
private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER SelectorProvider.provider();public NioServerSocketChannel() {this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}public NioServerSocketChannel(ServerSocketChannel channel) {super(null, channel, SelectionKey.OP_ACCEPT);config new NioServerSocketChannelConfig(this, javaChannel().socket());
}DEFAULT_SELECTOR_PROVIDER是Provider对象用于创建通道和选择器newSocket方法返回一个ServerSocketChannel对象如下所示:
private static ServerSocketChannel newSocket(SelectorProvider provider) {try {return provider.openServerSocketChannel();} catch (IOException e) {throw new ChannelException(Failed to open a server socket., e);}
}NioServerSocketChannel中还维护了一个config对象用于储存该通道相关的配置后续通过通道对象的config()方法获取该config对象。 继续调用父类的构造方法:
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {super(parent);this.ch ch;this.readInterestOp readInterestOp;try {ch.configureBlocking(false);} catch (IOException e) {try {ch.close();} catch (IOException e2) {logger.warn(Failed to close a partially initialized socket., e2);}throw new ChannelException(Failed to enter non-blocking mode., e);}
}// super(parent)内容如下:
protected AbstractChannel(Channel parent) {this.parent parent;id newId();unsafe newUnsafe();pipeline newChannelPipeline();
}因此NioServerSocketChannel中包含如下属性: [1] SelectableChannel ch实际为ServerSocketChannel类型即NIO中的服务端通道类型并将其配置为非阻塞类型以便后续向选择器注册 [2] int readInterestOp: 值固定为SelectionKey.OP_ACCEPT表示仅处理连接事件; [3] pipeline: Netty的Pipeline组件每个channel都有一个属于自己的Pipeline对象; [4] unsafe: 对底层IO进行了封装实际的读写操作在该类中进行处理; [5] 其他: id唯一ID标识parent固定为空。
1.2 NioServerSocketChannel注册 NioServerSocketChannel包含了ServerSocketChannel对象向选择器注册NioServerSocketChannel本质是将ServerSocketChannel注册到选择器 在Netty启动流程流程中依次构造ServerSocketChannel, 并注册到选择器上具体逻辑为:
// NioServerSocketChannel的父类AbstractNioChannel中
// 删除try-catch异常逻辑
protected void doRegister() throws Exception {boolean selected false;for (;;) {selectionKey javaChannel().register(eventLoop().unwrappedSelector(), 0, this);return;}
}其中: javaChannel()获取NioServerSocketChannel对象的ServerSocketChannel属性eventLoop().unwrappedSelector()为NioEventLoop这个线程绑定的选择器此处的this表明将ServerSocketChannel注册到选择器上时将当前的NioServerSocketChannel对象作为attachment保存到SelectionKey中并使用volatile SelectionKey selectionKey;属性保存了注册结果。
说明后续选择器会执行select而阻塞当该选择器被IO事件唤醒时可通过SelectionKey的attachment获取NioServerSocketChannel对象从而可以获取包括ServerSocketChannel、Pipeline、Config等其他所有相关信息。
1.3 NioServerSocketChannel处理连接
章节1.1中提到了NioServerSocketChannel的unsafe属性unsafe用于封装底层具体的IO行为具体的实现类为NioMessageUnsafe.
当有连接请求到达NioServerSocketChannel后进入NioMessageUnsafe的read()方法中(详细的调用流程和线程处理关系在后续Netty的消息处理流程中介绍, 这里仅对read方法实现逻辑进行说明)read方法省去内存分配优化策略以及异常处理逻辑后的主线逻辑如下:
private final class NioMessageUnsafe extends AbstractNioUnsafe {private final ListObject readBuf new ArrayListObject();Overridepublic void read() {// ...final ChannelPipeline pipeline pipeline();do {// ...doReadMessages(readBuf);} while (allocHandle.continueReading());int size readBuf.size();for (int i 0; i size; i ) {readPending false;pipeline.fireChannelRead(readBuf.get(i));}readBuf.clear();pipeline.fireChannelReadComplete();}
}readBuf是一个列表类型用于存放解析后的消息对象解析完成后依次遍历readBuf并调用pipeline.fireChannelRead将消息对象发送至Netty的Pipeline组件(后面单独介绍)。
解析逻辑在doReadMessages方法中:
protected int doReadMessages(ListObject buf) throws Exception {SocketChannel ch SocketUtils.accept(javaChannel());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;
}// SocketUtils.accept(javaChannel())代码逻辑:
public static SocketChannel accept(final ServerSocketChannel serverSocketChannel) throws IOException {// 删除try-catch异常逻辑return AccessController.doPrivileged(new PrivilegedExceptionActionSocketChannel() {Overridepublic SocketChannel run() throws IOException {return serverSocketChannel.accept();}});
}javaChannel()得到ServerSocketChannel对象serverSocketChannel.accept()得到客户端通道对象SocketChannel。将当前服务端通道NioServerSocketChannel对象和得到的客户端通道对象SocketChannel作为参数构造NioSocketChannel对象。
2.NioSocketChannel
与NioServerSocketChannel相似NioSocketChannel也是Netty对NIO中ServerSocketChannel的封装和增强。本章节内容将包含SocketChannel的构建、配置、向选择器注册以及读取数据如下所示:
// 得到SocketChannel对象
SocketChannel socketChannel serverSocketChannel.accept();// SocketChannel通道设置为非阻塞
socketChannel.configureBlocking(false);// 将SocketChannel通道注册至选择器
socketChannel.register(Selector, opts, attachment);// 从SocketChannel通道读取数据值缓冲区
socketChannel.read(ByteBuffer)2.1 NioSocketChannel构造函数 每个客户端连接对应一个通道即一个NioSocketChannel对象。 Netty收到客户端连接时会调用NioSocketChannel构造函数创建通道对象如下所示:
public NioSocketChannel(Channel parent, SocketChannel socket) {super(parent, socket);config new NioSocketChannelConfig(this, socket.socket());
}parent为NioServerSocketChannel对象socket为NIO中SocketChannel对象。NioSocketChannel与NioServerSocketChannel相似维持了一个config配置类用于存放和读取通道的配置信息。 继续沿着super调用父类的构造方法:
protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {super(parent, ch, SelectionKey.OP_READ);
}protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {super(parent);this.ch ch;this.readInterestOp readInterestOp;try {ch.configureBlocking(false);} catch (IOException e) {try {ch.close();} catch (IOException e2) {logger.warn(Failed to close a partially initialized socket., e2);}throw new ChannelException(Failed to enter non-blocking mode., e);}
}protected AbstractChannel(Channel parent) {this.parent parent;id newId();unsafe newUnsafe();pipeline newChannelPipeline();
}上述构造过程逻辑较为简单为NioSocketChannel创建一个Unsafe对象和Pipeline对象以及将ch属性即SocketChannel设置为非阻塞。
2.2 注册选择器
NioServerSocketChannel接收客户端连接构造出NioSocketChannel对象并通过Pipeline.fireChannelRead触发Inbound读事件后通过Pipiline进入ServerBootstrapAcceptor处理器的channelRead方法:
public void channelRead(ChannelHandlerContext ctx, Object msg) {final Channel child (Channel) msg;// ...childGroup.register(child).addListener(new ChannelFutureListener() {//...});
}由章节1可知msg消息为NioSocketChannelchildGroup为线程池NioEventLoopGroup对象(workgroup)。 childGroup.register(child)表示将NioSocketChannel注册到workgroup的一个线程中经过Unsafe对象最终会进入NioSocketChannel的doRegister方法:
Override
protected void doRegister() throws Exception {// ...selectionKey javaChannel().register(eventLoop().unwrappedSelector(), 0, this);// ...
}javaChannel()为NioSocketChannel的ch属性即SocketChannel通道对象eventLoop().unwrappedSelector()为选择器this为NioSocketChannel对象本身返回的SelectionKey也作为属性保存在NioSocketChannel类中。 说明后续选择器会执行select而阻塞当有可读消息到达时被唤醒。可通过SelectionKey得到NioSocketChannel对象从而得到相关的SocketChannel、Pipeline、Config等其他所有相关信息。
2.3 读取消息
当有可读时间到达时NioEvetLoop会从阻塞中被唤醒从而执行processSelectedKeys处理IO事件
private void processSelectedKeys() {// ...processSelectedKeysOptimized();// ...
}private void processSelectedKeysOptimized() {for (int i 0; i selectedKeys.size; i) {final SelectionKey k selectedKeys.keys[i];selectedKeys.keys[i] null;final Object a k.attachment();processSelectedKey(k, (AbstractNioChannel) a);}
}遍历已就绪的IO事件调用processSelectedKey方法处理此时k为NIO的SelectionKey对象而attachment为NioSocketChannel对象。
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {final AbstractNioChannel.NioUnsafe unsafe ch.unsafe();int readyOps k.readyOps();//...if ((readyOps (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) ! 0 || readyOps 0) {unsafe.read();}// ...
}根据SelectionKey和NioSocketChannel对象的readyOps确定此时IO事件为可读消息进入unsafe.read()
Override
public final void read() {final ChannelConfig config config();final ChannelPipeline pipeline pipeline();final ByteBufAllocator allocator config.getAllocator();ByteBuf byteBuf null;boolean close false;// ...do {// ...// 1.分配ButeBuf缓冲对象byteBuf allocHandle.allocate(allocator);// 2.将数据读取到ButeBuf缓冲对象allocHandle.lastBytesRead(doReadBytes(byteBuf));if (allocHandle.lastBytesRead() 0) {byteBuf.release();byteBuf null;break;}readPending false;// 3.向Pipeline传递可读消息pipeline.fireChannelRead(byteBuf);byteBuf null;// 直到读取完所有消息内容} while (allocHandle.continueReading());// ...// 触发消息读取完成事件pipeline.fireChannelReadComplete();// ...
}代码较为清晰重点包含3个步骤创建ByteBuf缓冲对象(Netty自定义的而非NIO的ByteBuffer); 将消息读取到ButeBuf对象向Pipeline触发可读事件(在Pipeline的Handler中传递并处理消息)其中核心逻辑在于doReadBytes(byteBuf):
Override
protected int doReadBytes(ByteBuf byteBuf) throws Exception {// ...return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}javaChannel()是NIO的SocketChannel对象继续跟进ByteBuf的writeBytes方法进入:
Override
public int writeBytes(ScatteringByteChannel in, int length) throws IOException {//...int writtenBytes setBytes(writerIndex, in, length);//...return writtenBytes;
}Override
public final int setBytes(int index, ScatteringByteChannel in, int length) throws IOException {try {return in.read(internalNioBuffer(index, length));} catch (ClosedChannelException ignored) {return -1;}
}可以看到底层逻辑在于in.read(internalNioBuffer(index, length)), 返回一个ByteBuffer对象in此时为SocketChannel, 即本质是调用NIO通道的API将数据读取至缓冲区: SocketChannel.read(ByteBuffer).
2.3 响应消息
Netty中Pipeline的任何一个Handler中都可以发送响应消息响应消息也会沿着Pipeline的流水线传递并经过网卡传递出去:
Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {ctx.writeAndFlush(hello);
}注意需要在此Handler前添加StringEncoder编码器将String类型转为ByteBuf类型否则会抛出异常。因为NioSocketChannel的Unsafe对象也维持在了Pipeline的HeadContext对象中所有的消息最终会经过Unsafe的write方法而Unsafe只会处理ByteBuf类型消息其他类型会抛出异常。
追踪ctx.writeAndFlush(hello)进入invokeWriteAndFlush方法:
void invokeWriteAndFlush(Object msg, ChannelPromise promise) {// ...invokeWrite0(msg, promise);invokeFlush0();// ...
}依次调用invokeWrite0和invokeFlush0实现写操作和刷盘操作, 分别进入Unsafe对象的write和flush方法:
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {unsafe.write(msg, promise);
}public void flush(ChannelHandlerContext ctx) {unsafe.flush();
}unsafe最终调用doWrite方法实现IO功能:
protected void doWrite(ChannelOutboundBuffer in) throws Exception {SocketChannel ch javaChannel();int writeSpinCount config().getWriteSpinCount();do {// ... ByteBuffer buffer nioBuffers[0];int attemptedBytes buffer.remaining();final int localWrittenBytes ch.write(buffer);--writeSpinCount;// ... } while (writeSpinCount 0);incompleteWrite(writeSpinCount 0);
}核心逻辑在与ch.write(buffer)其中ch和buffer分别是NIO的SocketChannel和ByteBuffer, 即Netty向客户端发送消息底层仍是借助NIO的API.
