Netty源码看这篇就够了

前言

后面打算开始撸其他框架源码，而Netty对Java NIO的一层封装，提供了一套简单易用的API，经常被其他框架拿来用，我先花了点时间研究了下。这里整理下对源码的解读，以及对几个关键对象的介绍。分析了之前两篇流水账式的源码分析的不足，这次尝试聚焦几个不同重点进行分析。

个人netty注释版本：https://gitee.com/Nortyr/netty

原netty地址：https://github.com/netty/netty

看完能收获什么

Java网络编程介绍

一个简单的EchoServerDemo

Bootstrap

Channel

ChannelPipeline & ChannelHandler

EventLoopGroup & EventLoop

ChannelFuture

Java网络编程介绍

BIO模型。庞大的线程消耗，消费消息如果很漫长，这个服务就是个灾难。

public void server(int port) throws IOException{

final ServerSocket socket=new ServerSocket(port);

for (;;){

//接受连接

final Socket clientSock = socket.accept();

System.out.println("Accept connection from"+ clientSock);

//创建一个线程来处理连接

new Thread(new Runnable() {

@Override

public void run() {

OutputStream out;

try {

out=clientSock.getOutputStream();

... doSomeThing...

//将消息写给已连接的客户端

out.write("Hello world".getBytes(Charset.forName("UTF-8")));

out.flush();

clientSock.close();

}

...略...

}

}).start();

}

}

故此Java设计出了NIO，这里找到个Doug Lea大神的一篇 NIOppt http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf 感兴趣的可以看一下，十分精简。下面这部分会结合这个ppt进行讲解，看完这个ppt的可以直接略过到下一个部分。

public static void main(String[] args) throws IOException {

//创建ServerSocketChannel，处理接入连接

ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();

//创建Selector

Selector selector=Selector.open();

//设置是否为非阻塞

serverSocketChannel.configureBlocking(false);

//创建注册channel进selector的创立连接时间

serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

//绑定端口号

serverSocketChannel.socket().bind(new InetSocketAddress(8080));

while (true){

if(serverSocketChannel.isOpen()){

// 通过 Selector 选择 Channel

int selectNums = selector.select(1000L);

if (selectNums == 0) {

continue;

}

// 遍历可选择的 Channel 的 SelectionKey 集合

for (SelectionKey selectKey:selector.selectedKeys()) {

// 忽略无效的 SelectionKey

if (!selectKey.isValid()) {

continue;

}

//新建立的连接

if(selectKey.isAcceptable()){

//获取新连接创建的channel

SocketChannel socketChannel= ((ServerSocketChannel) selectKey.channel()).accept();

if(socketChannel!=null){

//设置为非阻塞

socketChannel.configureBlocking(false);

//注册进selector

socketChannel.register(selector,SelectionKey.OP_READ);

}

}

//处理读时间

if(selectKey.isReadable()){

SocketChannel socketChannel= (SocketChannel) selectKey.channel();

if(socketChannel!=null){

//读取数据

ByteBuffer buffer = ByteBuffer.allocate(1024);

int bytesRead = socketChannel.read(buffer);

if(bytesRead==-1){

socketChannel.register(selector,0);

socketChannel.close();

}else{

buffer.flip();

byte[] bytes = new byte[buffer.remaining()];

System.arraycopy(buffer.array(), buffer.position(), bytes, 0, buffer.remaining());

System.out.println(new String(bytes, "UTF-8"));

}

}

}

}

}

}

}

Java NIO的几个核心api

Channels

与支持非阻塞读取的文件，socket等建立连接。

Buffers

本质是一块内存，用于和NIO通道进行交互。

Selectors

把Channel和需要的事件注册到Selector上面，告诉一组channel中的哪一个有IO事件。

SelectionKeys

维护IO事件状态和绑定

几个核心api的关系

Channel和Buffer

2个交互关系如图所示

Selector/Channel/SelectionKey

一个Selector可以监听多个Channel

一个Selector和Channel的绑定关系为SelectionKey

Channel

这里我们以NioServerSocketChannel为例，看一下Channel

public interface Channel extends AttributeMap, ChannelOutboundInvoker, Comparable {

ChannelId id();

EventLoop eventLoop();

Channel parent();

ChannelConfig config();

boolean isOpen();

boolean isRegistered();

boolean isActive();

ChannelMetadata metadata();

SocketAddress localAddress();

SocketAddress remoteAddress();

ChannelFuture closeFuture();

boolean isWritable();

long bytesBeforeUnwritable();

long bytesBeforeWritable();

Unsafe unsafe();

ChannelPipeline pipeline();

ByteBufAllocator alloc();

@Override

Channel read();

@Override

Channel flush();

/**

* 调用Javanio方法封装

*/

interface Unsafe {

RecvByteBufAllocator.Handle recvBufAllocHandle();

/**

* 返回地址

*/

SocketAddress localAddress();

/**

* 返回远程地址

*/

SocketAddress remoteAddress();

/**

* 注册Channel，注册完成后通知ChannelPromise

*/

void register(EventLoop eventLoop, ChannelPromise promise);

/**

* 将ip地址绑定到Channel，完成后通知ChannelPromise

*/

void bind(SocketAddress localAddress, ChannelPromise promise);

/**

* 连接远程ip地址

*/

void connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);

/**

* 断开连接，完成后通知ChannelPromise

*/

void disconnect(ChannelPromise promise);

/**

* 关闭channel，通知ChannelPromise

*/

void close(ChannelPromise promise);

/**

* 关闭，不处罚任何事件

*/

void closeForcibly();

/**

* 注销channel，通知ChannelPromise

*/

void deregister(ChannelPromise promise);

/**

* 调用读取操作

*/

void beginRead();

/**

* 调用写操作

*/

void write(Object msg, ChannelPromise promise);

/**

* 清空所有通过ChannelPromise预定的写操作

*/

void flush();

ChannelPromise voidPromise();

/**

* 返回存储待处理写入请求的Channel的ChannelOutboundBuffer。

*/

ChannelOutboundBuffer outboundBuffer();

}

}

public abstract class AbstractNioChannel extends AbstractChannel {

private final SelectableChannel ch;

protected final int readInterestOp;

volatile SelectionKey selectionKey;

boolean readPending;

private final Runnable clearReadPendingRunnable = new Runnable() {

@Override

public void run() {

clearReadPending0();

}

};

private ChannelPromise connectPromise;

private Future connectTimeoutFuture;

private SocketAddress requestedRemoteAddress;

}

以上是Channel接口和AbstractNioChannel的抽象类，这里给大家精简了下，从Channel定义的各个方法可以看出，netty的Channel是对原始Channel的一层封装。其中所有的nio的操作封装在了Unsafe中，并进行了一定的增强，例如回调之类的。从AbstractNioChannel可以更加直观的看出，netty对Channel SelectionKey的封装，并添加了自己的回调ChannelPromise从而使方法更加易于使用。

ChannelPipeline & ChannelHandler

基础讲解

ChannelPipeline的初始化

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {

...其余略...

private final DefaultChannelPipeline pipeline;

protected AbstractChannel(Channel parent) {

this.parent = parent;

id = newId();

unsafe = newUnsafe();

pipeline = newChannelPipeline();

}

protected DefaultChannelPipeline newChannelPipeline() {

return new DefaultChannelPipeline(this);

}

}

ChannelPipeline内部结构概述

public class DefaultChannelPipeline implements ChannelPipeline {

final AbstractChannelHandlerContext head;

final AbstractChannelHandlerContext tail;

private final Channel channel;

private final ChannelFuture succeededFuture;

protected DefaultChannelPipeline(Channel channel) {

this.channel = ObjectUtil.checkNotNull(channel, "channel");

succeededFuture = new SucceededChannelFuture(channel, null);

voidPromise = new VoidChannelPromise(channel, true);

tail = new TailContext(this);

head = new HeadContext(this);

head.next = tail;

tail.prev = head;

}

...略

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {

TailContext(DefaultChannelPipeline pipeline) {

super(pipeline, null, TAIL_NAME, TailContext.class);

setAddComplete();

}

... 略

}

final class HeadContext extends AbstractChannelHandlerContext

implements ChannelOutboundHandler, ChannelInboundHandler {

private final Unsafe unsafe;

HeadContext(DefaultChannelPipeline pipeline) {

super(pipeline, null, HEAD_NAME, HeadContext.class);

unsafe = pipeline.channel().unsafe();

setAddComplete();

}

...略

}

}

上面列举了ChannelPipeline的创建，以及ChannelPipeline的内部结构。可以看出它维护了一个双向链表。我们在添加handler的时候就是往这个链表中添加的。

ChannelHandler 添加进ChannelPipeline后会被封装成ChannelHandlerContext,会判断是ChannelInboundHandler还是ChannelOutboundHandler的子类，对inbound和outbound这两个属性进行赋值，ChannelInboundHandler的子类inbound为true，outbound为false，ChannelOutboundHandler反之。ChannelPipeline内部调用方法时，会使用fireXXXXX()的方法，会利用责任链模式进行调用，这时候会用到这个属性进行判断，是否有对应方法，从而进行调用（后面会详细讲解下）。

final class DefaultChannelHandlerContext extends AbstractChannelHandlerContext {

DefaultChannelHandlerContext(

DefaultChannelPipeline pipeline, EventExecutor executor, String name, ChannelHandler handler) {

super(pipeline, executor, name, isInbound(handler), isOutbound(handler));

if (handler == null) {

throw new NullPointerException("handler");

}

this.handler = handler;

}

private static boolean isInbound(ChannelHandler handler) {

return handler instanceof ChannelInboundHandler;

}

private static boolean isOutbound(ChannelHandler handler) {

return handler instanceof ChannelOutboundHandler;

}

private AbstractChannelHandlerContext findContextInbound() {

AbstractChannelHandlerContext ctx = this;

do {

ctx = ctx.next;

} while (!ctx.inbound);

return ctx;

}

private AbstractChannelHandlerContext findContextOutbound() {

AbstractChannelHandlerContext ctx = this;

do {

ctx = ctx.prev;

} while (!ctx.outbound);

return ctx;

}

}

方法调用

这里就用了责任链的方式调用方法，确定下一个调用哪一个节点，就是通过inbound outbound

这两个字段决定的。

public final void read() {

...省略掉部分无用代码

final ChannelConfig config = config();

final ChannelPipeline pipeline = pipeline();

ByteBuf byteBuf = null;

boolean close = false;

try {

do {

byteBuf = allocHandle.allocate(allocator);

pipeline.fireChannelRead(byteBuf);

byteBuf = null;

} while (allocHandle.continueReading());

pipeline.fireChannelReadComplete();

if (close) {

closeOnRead(pipeline);

}

}

}

@Override

public ChannelHandlerContext fireChannelRead(final Object msg) {

invokeChannelRead(findContextInbound(), msg);

return this;

}

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {

//如果msg实现了ReferenceCounted，进行特殊操作

final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);

EventExecutor executor = next.executor();

if (executor.inEventLoop()) {

next.invokeChannelRead(m);

} else {

executor.execute(new Runnable() {

@Override

public void run() {

next.invokeChannelRead(m);

}

});

}

}

private void invokeChannelRead(Object msg) {

if (invokeHandler()) {

try {

//调用下一个节点的channelRead方法

((ChannelInboundHandler) handler()).channelRead(this, msg);

} catch (Throwable t) {

invokeExceptionCaught(t);

}

} else {

fireChannelRead(msg);

}

}

下面就是调用你自定义的ChannelInboundHandler子类的覆盖方法了，这里就不过多赘述。

EventLoopGroup & EventLoop

初始化

EventLoopGroup初始化创建的时候，会创建对应数量的EventLoop，如果没有指定，默认创建cpu核心数量*2个EventLoop

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {

//默认线程数是cpu核心的2倍

static {

DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(

"io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));

if (logger.isDebugEnabled()) {

logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);

}

}

}

public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,

EventExecutorChooserFactory chooserFactory, Object... args) {

children = new EventExecutor[nThreads];

for (int i = 0; i < nThreads; i ++) {

boolean success = false;

try {

//创建对应数量的EventLoop

children[i] = newChild(executor, args);

success = true;

}

}

chooser = chooserFactory.newChooser(children);

final FutureListener