Netty5源码分析

实例

样例代码来自于io.netty.example.telnet.TelnetClient，完整样例请参考NettyExample工程。

客户端和服务端比较相似，所以本篇会在一定程度上略去重复的部分，以减少篇幅。

1public void run() throws Exception {
2    EventLoopGroup group = new NioEventLoopGroup();
3    try {
4        Bootstrap b = new Bootstrap();
5        b.group(group)
6         .channel(NioSocketChannel.class)
7         .handler(new TelnetClientInitializer());
8
9        // Start the connection attempt.
10        Channel ch = b.connect(host, port).sync().channel();
11
12        // Read commands from the stdin.
13        ChannelFuture lastWriteFuture = null;
14        BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
15        for (;;) {
16            String line = in.readLine();
17            if (line == null) {
18                break;
19            }
20
21            // Sends the received line to the server.
22            lastWriteFuture = ch.writeAndFlush(line + "\r\n");
23
24            // If user typed the 'bye' command, wait until the server closes
25            // the connection.
26            if ("bye".equals(line.toLowerCase())) {
27                ch.closeFuture().sync();
28                break;
29            }
30        }
31
32        // Wait until all messages are flushed before closing the channel.
33        if (lastWriteFuture != null) {
34            lastWriteFuture.sync();
35        }
36    } finally {
37        group.shutdownGracefully();
38    }
39}

客户端启动

1Bootstrap b = new Bootstrap();  //tag0
2b.group(group) //tag1
3.channel(NioSocketChannel.class) //tag2
4.handler(new TelnetClientInitializer());//tag3

tag0代码主要初始化了父类的 options和attrs 属性；代码略。

tag1设置了group属性

1@SuppressWarnings("unchecked")
2public B group(EventLoopGroup group) {
3    if (group == null) {
4        throw new NullPointerException("group");
5    }
6    if (this.group != null) {
7        throw new IllegalStateException("group set already");
8    }
9    this.group = group;
10    return (B) this;
11} 

tag2设置了channelFactory属性

1public Bootstrap channel(Class<? extends Channel> channelClass) {
2    if (channelClass == null) {
3        throw new NullPointerException("channelClass");
4    }
5    return channelFactory(new BootstrapChannelFactory<Channel>(channelClass));
6} 

tag3设置了handler属性

public B handler(ChannelHandler handler) { if (handler == null) { throw new NullPointerException("handler"); } this.handler = handler; return (B) this; }

下面开始第二段代码分析，依次执行下面的方法。

1Channel ch = b.connect(host, port) //tag4
2.sync().channel(); //tag5
3
4 public ChannelFuture connect(String inetHost, int inetPort) {
5    return connect(new InetSocketAddress(inetHost, inetPort));
6}
7
8public ChannelFuture connect(SocketAddress remoteAddress) {
9    if (remoteAddress == null) {
10        throw new NullPointerException("remoteAddress");
11    }
12
13    validate();
14    return doConnect(remoteAddress, localAddress());
15}
16
17 private ChannelFuture doConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
18    final ChannelFuture regFuture = initAndRegister();//tag4.1
19    final Channel channel = regFuture.channel();
20    if (regFuture.cause() != null) {
21        return regFuture;
22    }
23
24    final ChannelPromise promise = channel.newPromise();
25    if (regFuture.isDone()) {
26 doConnect0(regFuture, channel, remoteAddress, localAddress, promise);//tag4.2
27    } else {
28        regFuture.addListener(new ChannelFutureListener() {
29            @Override
30            public void operationComplete(ChannelFuture future) throws Exception {
31                doConnect0(regFuture, channel, remoteAddress, localAddress, promise);
32            }
33        });
34    }
35
36    return promise;
37}
38
39分析tag4.1代码，细心的读者注意到，这些和服务端的代码执行过程是一样的。运用模板模式，子类定义独特的实现。
40
41final ChannelFuture AbstractBootstrap.initAndRegister() {
42    Channel channel;
43    try {
44        channel = createChannel();//tag4.1.1
45
46    } catch (Throwable t) {
47        return VoidChannel.INSTANCE.newFailedFuture(t);
48    }
49
50    try {
51        init(channel);//tag4.1.2
52    } catch (Throwable t) {
53        channel.unsafe().closeForcibly();
54        return channel.newFailedFuture(t);
55    }
56
57    ChannelPromise regFuture = channel.newPromise();
58    channel.unsafe().register(regFuture);//tag4.1.3
59    if (regFuture.cause() != null) {
60        if (channel.isRegistered()) {
61            channel.close();
62        } else {
63            channel.unsafe().closeForcibly();
64        }
65    }

分析 tag4.1.1，里面通过反射来实例化NioSocketChannel

1    @Override
2Channel createChannel() {
3    EventLoop eventLoop = group().next();
4    return channelFactory().newChannel(eventLoop);//tag4.1.1.1
5
6}
7
8public NioSocketChannel(EventLoop eventLoop) {
9    this(eventLoop, newSocket());//调用下面的newSocket()方法
10}
11
12private static SocketChannel newSocket() {
13    try {
14        return SocketChannel.open();
15    } catch (IOException e) {
16        throw new ChannelException("Failed to open a socket.", e);
17    }
18}
19
20 public NioSocketChannel(EventLoop eventLoop, SocketChannel socket) {
21    this(null, eventLoop, socket);
22}
23
24protected AbstractNioByteChannel(Channel parent, EventLoop eventLoop, SelectableChannel ch) {
25    super(parent, eventLoop, ch, SelectionKey.OP_READ);//调用父类方法
26}
27
28protected AbstractNioChannel(Channel parent, EventLoop eventLoop, SelectableChannel ch, int readInterestOp) {
29    super(parent, eventLoop);//调用父类方法,tag4.1.1.1
30    this.ch = ch;
31    this.readInterestOp = readInterestOp;
32    try {
33        ch.configureBlocking(false);//tag4.1.1.2
34    } catch (IOException e) {
35        try {
36            ch.close();
37        } catch (IOException e2) {
38            if (logger.isWarnEnabled()) {
39                logger.warn(
40                        "Failed to close a partially initialized socket.", e2);
41            }
42        }
43
44        throw new ChannelException("Failed to enter non-blocking mode.", e);
45    }
46}
47
48protected AbstractChannel(Channel parent, EventLoop eventLoop) {
49    this.parent = parent;
50    this.eventLoop = validate(eventLoop);
51    unsafe = newUnsafe();
52    pipeline = new DefaultChannelPipeline(this);//tag4.1.1.1.1
53}

分析tag4.1.1.1.1，里面调用DefaultChannelPipeline构造器，和服务端的逻辑一样，故不作分析。

此时系统返回到tag4.1.1.2 继续执行ch.configureBlocking(false);，此时完成tag4.1.1 方法执行，开始执行tag4.1.2方法

1 @Override
2@SuppressWarnings("unchecked")
3void init(Channel channel) throws Exception {
4    ChannelPipeline p = channel.pipeline();
5    p.addLast(handler());//tag4.1.2.1
6
7    final Map<ChannelOption<?>, Object> options = options();
8    synchronized (options) {
9        for (Entry<ChannelOption<?>, Object> e: options.entrySet()) {
10            try {
11                if (!channel.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue()))  	{
12                    logger.warn("Unknown channel option: " + e);
13                }
14            } catch (Throwable t) {
15                logger.warn("Failed to set a channel option: " + channel, t);
16            }
17        }
18    }
19
20    final Map<AttributeKey<?>, Object> attrs = attrs();
21    synchronized (attrs) {
22        for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
23            channel.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
24        }
25    }
26}

分析tag4.1.2.1，里面将这个TelnetClientInitializer Handler加入到pipeline中，此时handler链是HeadHandler,TelnetClientInitializer,TailHandler,共计3个。

此时程序返回到tag4.1.3继续执行，

1@Override
2    public final void AbstractChannel.register(final ChannelPromise promise) {
3        if (eventLoop.inEventLoop()) {
4            register0(promise);
5        } else {
6            try {
7                eventLoop.execute(new Runnable() {
8                    @Override
9                    public void run() {
10                        register0(promise);//tag4.1.3.1
11                    }
12                });
13            } catch (Throwable t) {
14                logger.warn(
15                        "Force-closing a channel whose registration task was not accepted by an event loop: {}",
16                        AbstractChannel.this, t);
17                closeForcibly();
18                closeFuture.setClosed();
19                promise.setFailure(t);
20            }
21        }
22    }
23
24
25
26
27     private void AbstractChannel.AbstractUnsafe.register0(ChannelPromise promise) {
28        try {
29            // check if the channel is still open as it could be closed in the mean time when the register
30            // call was outside of the eventLoop
31            if (!ensureOpen(promise)) {
32                return;
33            }
34            doRegister();//tag4.1.3.1.1   
35            registered = true;
36            promise.setSuccess();
37            pipeline.fireChannelRegistered();//tag4.1.3.1.2
38            if (isActive()) {
39                pipeline.fireChannelActive();
40            }
41        } catch (Throwable t) {
42            // Close the channel directly to avoid FD leak.
43            closeForcibly();
44            closeFuture.setClosed();
45            if (!promise.tryFailure(t)) {
46                logger.warn(
47                        "Tried to fail the registration promise, but it is complete already. " +
48                                "Swallowing the cause of the registration failure:", t);
49            }
50        }
51    }

tag4.1.3.1.1 代码如下

1     @Override
2protected void AbstractNioChannel.doRegister() throws Exception {
3    boolean selected = false;
4    for (;;) {
5        try {
6            selectionKey = javaChannel().register(eventLoop().selector, 0, this);//tag4.1.3.1.2.1
7            return;
8        } catch (CancelledKeyException e) {
9            if (!selected) {
10                // Force the Selector to select now as the "canceled" SelectionKey may still be
11                // cached and not removed because no Select.select(..) operation was called yet.
12                eventLoop().selectNow();
13                selected = true;
14            } else {
15                // We forced a select operation on the selector before but the SelectionKey is still cached
16                // for whatever reason. JDK bug ?
17                throw e;
18            }
19        }
20    }
21}

tag4.1.3.1.2.1 把selector注册到javaChannel上；然后程序继续执行tag4.1.3.1.2代码。

1@Override
2public ChannelPipeline fireChannelRegistered() {
3    head.fireChannelRegistered();
4    return this;
5}
6
7@Override
8public ChannelHandlerContext fireChannelRegistered() {
9    DefaultChannelHandlerContext next = findContextInbound(MASK_CHANNEL_REGISTERED);
10    next.invoker.invokeChannelRegistered(next);
11    return this;
12}
13
14@Override
15@SuppressWarnings("unchecked")
16public final void ChannelInitializer.channelRegistered(ChannelHandlerContext ctx) throws Exception {
17    ChannelPipeline pipeline = ctx.pipeline();
18    boolean success = false;
19    try {
20        initChannel((C) ctx.channel());//tag4.1.3.1.2.1
21        pipeline.remove(this);//tag4.1.3.1.2.2
22        ctx.fireChannelRegistered();//tag4.1.3.1.2.3
23        success = true;
24    } catch (Throwable t) {
25        logger.warn("Failed to initialize a channel. Closing: " + ctx.channel(), t);
26    } finally {
27        if (pipeline.context(this) != null) {
28            pipeline.remove(this);
29        }
30        if (!success) {
31            ctx.close();
32        }
33    }
34}

后面的逻辑和服务端类似，此时执行的handler是TelnetClientInitializer，并执行ChannelInitializer的channelRegistered方法，channelRegistered方法里面接着调用了initChannel。

标记 tag4.1.3.1.2.1 代码如下

1@Override
2public void TelnetClientInitializer.initChannel(SocketChannel ch) throws Exception {
3    ChannelPipeline pipeline = ch.pipeline();
4
5    // Add the text line codec combination first,
6    pipeline.addLast("framer", new DelimiterBasedFrameDecoder(
7            8192, Delimiters.lineDelimiter()));
8    pipeline.addLast("decoder", DECODER);
9    pipeline.addLast("encoder", ENCODER);
10
11    // and then business logic.
12    pipeline.addLast("handler", CLIENTHANDLER);
13}

在完成tag4.1.3.1.2.2的 pipeline.remove(this);后，此时handler链如下：HeadHandler，DelimiterBasedFrameDecoder，StringDecoder，StringEncoder，TelnetClientHandler， TailHandler。

接着程序又开始执行下一个handler，最终找到TailHandler的channelRegistered方法。TailHandler的channelRegistered方法是空方法。

此时 tag4.1 的代码执行结束，开始执行 tag4.2的代码

private static void doConnect0( final ChannelFuture regFuture, final Channel channel, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {

1    // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
2    // the pipeline in its channelRegistered() implementation.
3    channel.eventLoop().execute(new Runnable() {
4        @Override
5        public void run() {
6            if (regFuture.isSuccess()) {
7                if (localAddress == null) {
8                    channel.connect(remoteAddress, promise);//tag4.2.1
9                } else {
10                    channel.connect(remoteAddress, localAddress, promise);
11                }
12                promise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
13            } else {
14                promise.setFailure(regFuture.cause());
15            }
16        }
17    });
18}
19
20@Override
21public ChannelFuture AbstractChannel.connect(SocketAddress remoteAddress, ChannelPromise promise) {
22    return pipeline.connect(remoteAddress, promise);
23}

经过一番计算，找到HeadHandler，执行unsafe的方法。

1@Override
2    public void HeadHandler.connect(
3            ChannelHandlerContext ctx,
4            SocketAddress remoteAddress, SocketAddress localAddress,
5            ChannelPromise promise) throws Exception {
6        unsafe.connect(remoteAddress, localAddress, promise);
7    }

AbstractNioChannel.AbstractNioUnsafe的 connect方法如下：

1  @Override
2    public void connect(
3            final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
4        if (!ensureOpen(promise)) {
5            return;
6        }
7
8        try {
9            if (connectPromise != null) {
10                throw new IllegalStateException("connection attempt already made");
11            }
12
13            boolean wasActive = isActive();
14            if (doConnect(remoteAddress, localAddress)) {//tag4.2.1.1
15                fulfillConnectPromise(promise, wasActive);//tag4.2.1.2
16            } else {
17                connectPromise = promise;
18                requestedRemoteAddress = remoteAddress;
19
20                // Schedule connect timeout.
21                int connectTimeoutMillis = config().getConnectTimeoutMillis();
22                if (connectTimeoutMillis > 0) {
23                    connectTimeoutFuture = eventLoop().schedule(new Runnable() {
24                        @Override
25                        public void run() {//tag4.2.1.3
26                            ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
27                            ConnectTimeoutException cause =
28                                    new ConnectTimeoutException("connection timed out: " + remoteAddress);
29                            if (connectPromise != null && connectPromise.tryFailure(cause)) {
30                                close(voidPromise());
31                            }
32                        }
33                    }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
34                }
35
36                promise.addListener(new ChannelFutureListener() {
37                    @Override
38                    public void operationComplete(ChannelFuture future) throws Exception {
39                        if (future.isCancelled()) {
40                            if (connectTimeoutFuture != null) {
41                                connectTimeoutFuture.cancel(false);
42                            }
43                            connectPromise = null;
44                            close(voidPromise());
45                        }
46                    }
47                });
48            }
49        } catch (Throwable t) {
50            if (t instanceof ConnectException) {
51                Throwable newT = new ConnectException(t.getMessage() + ": " + remoteAddress);
52                newT.setStackTrace(t.getStackTrace());
53                t = newT;
54            }
55            promise.tryFailure(t);
56            closeIfClosed();
57        }
58    }

tag4.2.1.1 代码如下，进行了bind本地端口和connect远程服务器的操作。

1@Override
2protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
3    if (localAddress != null) {
4        javaChannel().socket().bind(localAddress);
5    }
6
7    boolean success = false;
8    try {
9        boolean connected = javaChannel().connect(remoteAddress);//tag4.2.1.1.1
10        if (!connected) {
11            selectionKey().interestOps(SelectionKey.OP_CONNECT);//tag4.2.1.1.2
12        }
13        success = true;
14        return connected;
15    } finally {
16        if (!success) {
17            doClose();
18        }
19    }
20}       

tag4.2.1.1.2 里面执行了connect远程服务器的操作，，我的机器上该方法返回false（返回值详见connect方法说明）。然后会触发执行selectionKey().interestOps(SelectionKey.OP_CONNECT);

需要额外说明的是，此时服务器会触发channelActivi事件。在本例的服务端代码里，会在客户端连接时，发送消息给客户端。不过先暂时忽略服务端和客户端的数据交互，下文分析。

然后tag4.2.1.1 执行结束，由于此时的返回值是false，所以不会执行tag4.2.1.2的 fulfillConnectPromise(promise, wasActive);

然后程序继续执行tag4.2.1.3 代码，进行连接超时处理：如果设置了超时时间，那么等待指定的超时时间后，再看看是否已经连接上。如果连不上，则设置失败状态。

接着开始下一个事件循环，由于在tag4.2.1.1.2执行了selectionKey().interestOps(SelectionKey.OP_CONNECT)操作，会进入到下面的代码。这里我们重点关注tag4.3的代码。

1private static void NioEventLoop.processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
2 	//略XXXX
3              if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
4            // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
5            // See https://github.com/netty/netty/issues/924
6            int ops = k.interestOps();
7            ops &= ~SelectionKey.OP_CONNECT;
8            k.interestOps(ops);
9
10            unsafe.finishConnect();//tag4.3
11    
12    //略XXXX
13
14}
15
16@Override
17    public void finishConnect() {
18        // Note this method is invoked by the event loop only if the connection attempt was
19        // neither cancelled nor timed out.
20
21        assert eventLoop().inEventLoop();
22        assert connectPromise != null;
23
24        try {
25            boolean wasActive = isActive();
26            doFinishConnect();//tag4.3.1
27            fulfillConnectPromise(connectPromise, wasActive);//tag4.3.2
28        } catch (Throwable t) {
29            if (t instanceof ConnectException) {
30                Throwable newT = new ConnectException(t.getMessage() + ": " + requestedRemoteAddress);
31                newT.setStackTrace(t.getStackTrace());
32                t = newT;
33            }
34
35            // Use tryFailure() instead of setFailure() to avoid the race against cancel().
36            connectPromise.tryFailure(t);
37            closeIfClosed();
38        } finally {
39            // Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
40            // See https://github.com/netty/netty/issues/1770
41            if (connectTimeoutFuture != null) {
42                connectTimeoutFuture.cancel(false);
43            }
44            connectPromise = null;
45        }
46    }
47
48    @Override
49protected void NioSocketChannel.doFinishConnect() throws Exception {
50    if (!javaChannel().finishConnect()) {
51        throw new Error();
52    }
53}

在执行完下面的boolean promiseSet = promise.trySuccess(); 方法后，实例代码中的 Channel ch = b.connect(host, port).sync().channel();就执行完毕了，然后主线程就阻塞在实例代码中的 String line = in.readLine();这句代码里了。

1private void AbstractNioChannel.AbstractNioUnsafe.fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
2        // trySuccess() will return false if a user cancelled the connection attempt.
3        boolean promiseSet = promise.trySuccess();
4
5        // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
6        // because what happened is what happened.
7        if (!wasActive && isActive()) {
8            pipeline().fireChannelActive();//tag4.3.2.1
9        }
10
11        // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
12        if (!promiseSet) {
13            close(voidPromise());
14        }
15    }
16    
17 
18@Override
19public ChannelPipeline fireChannelActive() {
20    head.fireChannelActive();//tag4.3.2.1.1
21
22    if (channel.config().isAutoRead()) {
23        channel.read();//tag4.3.2.1.2
24    }
25
26    return this;
27}    

此时，继续执行tag4.3.2.1的代码，进而执行tag4.3.2.1.1的代码，最终执行TailHandler.channelActive方法。由于TailHandler类内部的方法基本都是空实现，所以不再贴代码了。然后再执行tag4.3.2.1.2的channel.read();代码，最终执行了AbstractNioChannel.doBeginRead()方法的selectionKey.interestOps(interestOps | readInterestOp);，等同于执行了selectionKey.interestOps(SelectionKey.OP_READ);。

此时方法返回，在NioEventLoop.run()经过了一些简单的数据清理后，然后有机会对服务端的channelActive时发送的数据进行处理了(在tag4.2.1.1.2曾经提过)。客户端和服务端交互过程详见下篇。