Netty Channel分类与新链接接入

Netty新链接接入流程

1.检测新连接->2.建立NioSocketChannel->3.分配线程及注册selector->4.向selector注册读事件

检测新链接

步骤

• processSelectedKey(key,channel)入口
  • NioMessageUnsafe.read()
    • doReadMessages() while循环
      • javaChannel().accept()

分析

• processSelectedKey(key,channel)入口

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        try {
            int readyOps = k.readyOps();
            // 省略代码...
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                // 调用与ServerSocketChannel绑定的
                unsafe.read();
            }
        } catch (CancelledKeyException ignored) {
            unsafe.close(unsafe.voidPromise());
        }
    }
复制代码

doReadMessages()

@Override
    protected int doReadMessages(List<Object> buf) throws Exception {
        // 接收客户端链接
        SocketChannel ch = SocketUtils.accept(javaChannel());
        try {
            if (ch != null) {
                // netty建立本身的客户端channel
                buf.add(new NioSocketChannel(this, ch));
                return 1;
            }
        } catch (Throwable t) {
            //省略代码...
        }

        return 0;
    }
复制代码

建立NioSocketChannel

步骤

• new NioSocketChannel(parent,ch) 入口
  • AbstractNioByteChannel(p,ch,op_read)
    • configureBlocking(false)&save op
    • create id,unsafe,pipeline
  • new NioSocketChannelConfig()
    • setTcpNoDelay(true)

分析

• new NioSocketChannel(parent,ch) 入口

public NioSocketChannel(Channel parent, SocketChannel socket) {
        super(parent, socket);
        // 建立channel对应的config
        config = new NioSocketChannelConfig(this, socket.socket());
    }
    protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
        // 指定channel关注读事件
        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) {
            //省略代码..
        }
    }
复制代码

• 其余过程服务端channel建立相似，咱们说下这点setTcpNoDelay(true)

public SocketChannelConfig setTcpNoDelay(boolean tcpNoDelay) {
        try {
            // 若是不是安卓就true
            javaSocket.setTcpNoDelay(tcpNoDelay);
        } catch (SocketException e) {
            throw new ChannelException(e);
        }
        return this;
    }
    当开启nagle算法时，客户端首先发送大小为1字节的第一个分组，随后其它分组到达发送缓冲区，因为上一个分组的应答尚未收到，因此TCP会先缓存新来的这4个小分组，并将其从新分组，组成一个大小为8(2+3+1+2)字节的”较大的”小分组。当第一个小分组的应答收到后，客户端将这个8字节的分组发送。总共发送的报文段（分组）个数为2。当传输数据存在大量交互数据时，nagle算法能够有效减小网络中的报文段个数
    /**
     * Disable Nagle's algorithm for this connection. Written data * to the network is not buffered pending acknowledgement of * previously written data. */ @Native public final static int TCP_NODELAY = 0x0001; 复制代码

Netty中的Channel的分类

步骤

• NioServerSocketChannel
• NioSocketChannel

分析

• 一个简化的类图

以前建立Channel过程当中，客户端和服务端channel有公共成员变量。它们的类继承关系如图：

新链接NioEventLoop分配和selector注册

步骤

服务端Channel的pipeline构成 Head --> ServerBootstrapAcceptor --> tail

• ServerBootstrapAcceptor
  • 添加childHandler
  • 设置options和attrs
  • Chooser选择NioEventLopp并注册selector

分析

• 在服务端channel建立的时候会在pipeline中添加个ServerBootstrapAcceptor。

p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(final 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(
                            ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });
复制代码

• 建立ServerBootstrapAcceptor 过程最终要的一点就是初始化childGroup，由于一会要从中选择EventLoop的其中一个，并将channel注册上去
• 上面的doReadMessages对于链接事件，其实读到的是NioSocketChannel

int size = readBuf.size();
    for (int i = 0; i < size; i ++) {
        readPending = false;
        // 关键点在这，这句话的意思是  serverSocketChannel.pipeline().fireChannelRead(nioSocketChannel)
        pipeline.fireChannelRead(readBuf.get(i));
    }

    // serverAcceptor的channelRead方法 
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 省略代码 ... 
        // child 也就是咱们上面说的 nioSocketChannel
        final Channel child = (Channel) msg;
        // 那么这句话的意思就是 把nioSocketChannel注册到EventLoop的其中一个的selector上
        childGroup.register(child).addListener(...);
    }
复制代码

• NioSocketChannel注册过程

public ChannelFuture register(Channel channel) {
        // next()其实就是咱们以前在EventLoop中提到的EventLoop选择的问题，这块是个轮询详情请看EventLoop那篇文章
        return next().register(channel);
    }
    public ChannelFuture register(final ChannelPromise promise) {
        ObjectUtil.checkNotNull(promise, "promise");
        // 这块实际上是调用与客户端Channel所对应的unsafe
        promise.channel().unsafe().register(this, promise);
        return promise;
    }
    // 一直跟到最后的register.
    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                // 这块尚未监听具体的事件
                selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
                return;
            } catch (CancelledKeyException e) {
                ...
            }
        }
    }
复制代码

• 设置selector的interestOps

// readInterestOp上边设置的是OP_READ，也就是监听读事件
    protected void doBeginRead() throws Exception {
        ...
        final int interestOps = selectionKey.interestOps();
        if ((interestOps & readInterestOp) == 0) {
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }
复制代码