netty5笔记-总体流程分析5-客户端连接过程

港控/mmm° 2022-05-27 12:40 207阅读 0赞

前面几篇文章，我们从服务端的视角，分析了从启动到接收连接，到连接的read-write-close。现在我们开始切换到客户端的视角，看看客户端连接服务端的一些实现细节。

还是从snoop的example代码开始，见HttpSnoopClient（稍有修改）：

**\[java\]**  [view plain][]  [copy][view plain]

1.  public static void main(String\[\] args) throws Exception \{  
2.      // 配制client.  
3.      EventLoopGroup group = new NioEventLoopGroup();  
4.      try \{  
5.          Bootstrap b = new Bootstrap();  
6.          b.group(group)  
7.           .channel(NioSocketChannel.class)  
8.           .handler(new HttpSnoopClientInitializer(null));  
9.    
10.         // 创建连接.  
11.         Channel ch = b.connect("127.0.0.1", 80).sync().channel();  
12.   
13.         // 构造HTTP请求.  
14.         HttpRequest request = new DefaultFullHttpRequest(  
15.                 HttpVersion.HTTP\_1\_1, HttpMethod.GET, uri.getRawPath());  
16.         request.headers().set(HttpHeaderNames.HOST, host);  
17.         request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);  
18.         request.headers().set(HttpHeaderNames.ACCEPT\_ENCODING, HttpHeaderValues.GZIP);  
19.   
20.         // 发送HTTP请求.  
21.         ch.writeAndFlush(request);  
22.   
23.         // 等待服务端关闭连接  
24.         ch.closeFuture().sync();  
25.     \} finally \{  
26.         // 关闭group.  
27.         group.shutdownGracefully();  
28.     \}  
29. \}

有了前面ServerBootstrap的铺垫，这里就比较简单了，ServerBootstrap和Bootstrap都继承自AbstractBootstrap。但需要注意的是，在ServerBootstrap中group、channel、handler都是针对NioServerSocketChannel的设置，而切到Bootstrap后，group、channel、handler则是针对NioSocketChannel的设置了。相应的在ServerBootstrap中设置NioSocketChannel的属性和选项使用childAttr、childOption，而Bootstrap中设置NioSocketChannel则是直接使用attr、option。

下面直接进入主题，b.connect("127.0.0.1", 80)创建了一个客户端的链接，调用的方法：

**\[java\]**  [view plain][]  [copy][view plain]

1.      public ChannelFuture connect(String inetHost, int inetPort) \{  
2.          return connect(InetSocketAddress.createUnresolved(inetHost, inetPort));  
3.      \}  
4.    
5.      /\*\* 
6.       \* 创建一个连接. 
7.       \*/  
8.      public ChannelFuture connect(SocketAddress remoteAddress) \{  
9.          if (remoteAddress == null) \{  
10.             throw new NullPointerException("remoteAddress");  
11.         \}  
12.         // 主要验证必要的参数是否设置，如group何channelFactory(通过channel方法设置）  
13.         validate();  
14.         // 解析并连接  
15.         return doResolveAndConnect(remoteAddress, localAddress());  
16.     \}  
17.   
18.     private ChannelFuture doResolveAndConnect(SocketAddress remoteAddress, final SocketAddress localAddress) \{  
19.         // 初始化并注册到EventLoop（注册这块过程与server端类似，不再细讲）  
20.         final ChannelFuture regFuture = initAndRegister();  
21.         if (regFuture.cause() != null) \{  
22.             return regFuture;  
23.         \}  
24.   
25.         final Channel channel = regFuture.channel();  
26.         final EventLoop eventLoop = channel.eventLoop();  
27.         final NameResolver<SocketAddress> resolver = this.resolver.getResolver(eventLoop);  
28.   
29.         if (!resolver.isSupported(remoteAddress) || resolver.isResolved(remoteAddress)) \{  
30.             // Resolver has no idea about what to do with the specified remote address or it's resolved already.  
31.             return doConnect(remoteAddress, localAddress, regFuture, channel.newPromise());  
32.         \}  
33.         // 对remoteAddress进行解析，如果地址为ip(127.0.0.1)则直接返回，如地址为域名(www.baidu.com)则需要解析为ip(180.97.33.107)  
34.         final Future<SocketAddress> resolveFuture = resolver.resolve(remoteAddress);  
35.         final Throwable resolveFailureCause = resolveFuture.cause();  
36.   
37.         if (resolveFailureCause != null) \{  
38.             // Failed to resolve immediately  
39.             channel.close();  
40.             return channel.newFailedFuture(resolveFailureCause);  
41.         \}  
42.   
43.         // 在解析完成后调用doConnect(最终调用doConnect0)连接远程的端口  
44.         if (resolveFuture.isDone()) \{  
45.             // Succeeded to resolve immediately; cached? (or did a blocking lookup)  
46.             return doConnect(resolveFuture.getNow(), localAddress, regFuture, channel.newPromise());  
47.         \}  
48.   
49.         // Wait until the name resolution is finished.  
50.         final ChannelPromise connectPromise = channel.newPromise();  
51.         resolveFuture.addListener(new FutureListener<SocketAddress>() \{  
52.             @Override  
53.             public void operationComplete(Future<SocketAddress> future) throws Exception \{  
54.                 if (future.cause() != null) \{  
55.                     channel.close();  
56.                     connectPromise.setFailure(future.cause());  
57.                 \} else \{  
58.                     doConnect(future.getNow(), localAddress, regFuture, connectPromise);  
59.                 \}  
60.             \}  
61.         \});  
62.   
63.         return connectPromise;  
64.     \}  
65.       
66.     // 初始化pipeline、各个option和attr  
67.     void init(Channel channel) throws Exception \{  
68.         ChannelPipeline p = channel.pipeline();  
69.         p.addLast(handler());  
70.   
71.         final Map<ChannelOption<?>, Object> options = options();  
72.         synchronized (options) \{  
73.             for (Entry<ChannelOption<?>, Object> e: options.entrySet()) \{  
74.                 try \{  
75.                     if (!channel.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) \{  
76.                         logger.warn("Unknown channel option: " + e);  
77.                     \}  
78.                 \} catch (Throwable t) \{  
79.                     logger.warn("Failed to set a channel option: " + channel, t);  
80.                 \}  
81.             \}  
82.         \}  
83.   
84.         final Map<AttributeKey<?>, Object> attrs = attrs();  
85.         synchronized (attrs) \{  
86.             for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) \{  
87.                 channel.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());  
88.             \}  
89.         \}  
90.     \}   
91.   
92.     private static void doConnect0(  
93.             final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelFuture regFuture,  
94.             final ChannelPromise connectPromise) \{  
95.   
96.         // 这个方法在 channelRegistered()之前调用. 这样就给handlers机会在channelRegistered()方法中设置pipeline  
97.         final Channel channel = connectPromise.channel();  
98.         channel.eventLoop().execute(new Runnable() \{  
99.             @Override  
100.             public void run() \{  
101.                 if (regFuture.isSuccess()) \{  
102.                     // 执行channel的connect方法  
103.                     if (localAddress == null) \{  
104.                         channel.connect(remoteAddress, connectPromise);  
105.                     \} else \{  
106.                         channel.connect(remoteAddress, localAddress, connectPromise);  
107.                     \}  
108.                     connectPromise.addListener(ChannelFutureListener.CLOSE\_ON\_FAILURE);  
109.                 \} else \{  
110.                     connectPromise.setFailure(regFuture.cause());  
111.                 \}  
112.             \}  
113.         \});  
114.     \}

这个过程比较简单，1、创建一个NioSocketChannel实例，并用Bootstrap中的参数初始化该实例；2、将创建的channel注册到EventLoop中；3、创建一个解析host的任务(如果是ip，任务直接完成，如果是域名需要将域名解析为ip)；4、解析完成后，在EventLoop中调用channel.connect，连接到远程端口。

channel.connect将connect任务交给pipeline去处理，最终调用到TailContext中的connect方法，该方法调用代码：unsafe.connect(...)。

**\[java\]**  [view plain][]  [copy][view plain]

1.          // AbstractNioUnsafe  
2.          public final void connect(final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) \{  
3.              if (!promise.setUncancellable() || !ensureOpen(promise)) \{  
4.                  return;  
5.              \}  
6.    
7.              try \{  
8.                  if (connectPromise != null) \{  
9.                      throw new IllegalStateException("connection attempt already made");  
10.                 \}  
11.   
12.                 boolean wasActive = isActive();  
13.                 // 创建连接，如果连接立即成功了则返回true, 否则会返回false  
14.                 if (doConnect(remoteAddress, localAddress)) \{  
15.                     // 完成连接过程  
16.                     fulfillConnectPromise(promise, wasActive);  
17.                 \} else \{  
18.                     connectPromise = promise;  
19.                     requestedRemoteAddress = remoteAddress;  
20.   
21.                     // 如果设置了连接超时时间，则创建一个超时检测任务，如果超时未连接成功则关闭连接  
22.                     int connectTimeoutMillis = config().getConnectTimeoutMillis();  
23.                     if (connectTimeoutMillis > 0) \{  
24.                         connectTimeoutFuture = eventLoop().schedule(new OneTimeTask() \{  
25.                             @Override  
26.                             public void run() \{  
27.                                 ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;  
28.                                 ConnectTimeoutException cause =  
29.                                         new ConnectTimeoutException("connection timed out: " + remoteAddress);  
30.                                 if (connectPromise != null && connectPromise.tryFailure(cause)) \{  
31.                                     close(voidPromise());  
32.                                 \}  
33.                             \}  
34.                         \}, connectTimeoutMillis, TimeUnit.MILLISECONDS);  
35.                     \}  
36.                     // 如果连接被取消，则关闭前面创建的超时检测任务并关闭连接  
37.                     promise.addListener(new ChannelFutureListener() \{  
38.                         @Override  
39.                         public void operationComplete(ChannelFuture future) throws Exception \{  
40.                             if (future.isCancelled()) \{  
41.                                 if (connectTimeoutFuture != null) \{  
42.                                     connectTimeoutFuture.cancel(false);  
43.                                 \}  
44.                                 connectPromise = null;  
45.                                 close(voidPromise());  
46.                             \}  
47.                         \}  
48.                     \});  
49.                 \}  
50.             \} catch (Throwable t) \{  
51.                 promise.tryFailure(annotateConnectException(t, remoteAddress));  
52.                 closeIfClosed();  
53.             \}  
54.         \}  
55.   
56.         private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) \{  
57.             if (promise == null) \{  
58.                 // Closed via cancellation and the promise has been notified already.  
59.                 return;  
60.             \}  
61.   
62.             // 如果取消了连接则此处会返回false  
63.             boolean promiseSet = promise.trySuccess();  
64.   
65.             // 只要是连接确实打开了，则无论是否被取消channelActive都会触发  
66.             if (!wasActive && isActive()) \{  
67.                 pipeline().fireChannelActive();  
68.             \}  
69.   
70.             // 如果用户取消了连接，则会调用close方法，此方法会触发channelInactive  
71.             if (!promiseSet) \{  
72.                 close(voidPromise());  
73.             \}  
74.         \}  
75.   
76.     // doConnect来自NioSocketChannel  
77.     protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception \{  
78.         if (localAddress != null) \{  
79.             javaChannel().socket().bind(localAddress);  
80.         \}  
81.   
82.         boolean success = false;  
83.         try \{  
84.             // 发起连接请求，由于是非阻塞模式，因此主要是两种情况：1、连接本地，可能会立即完成，此时返回true;2、其他情况，没有立即完成连接，返回false  
85.             boolean connected = javaChannel().connect(remoteAddress);  
86.             if (!connected) \{  
87.                 // 注册OP\_CONNECT事件，如果连接成功则会进入NioEventLoop中的processSelectedKey方法  
88.                 selectionKey().interestOps(SelectionKey.OP\_CONNECT);  
89.             \}  
90.             success = true;  
91.             return connected;  
92.         \} finally \{  
93.             if (!success) \{  
94.                 doClose();  
95.             \}  
96.         \}  
97.     \}  
98.   
99.             // NioEventLoop中的processSelectedKey方法片段          
100.             if ((readyOps & SelectionKey.OP\_CONNECT) != 0) \{  
101.                 // 移除OP\_CONNECT否则会导致cpu 100%  
102.                 int ops = k.interestOps();  
103.                 ops &= ~SelectionKey.OP\_CONNECT;  
104.                 k.interestOps(ops);  
105.                 // 此调用最终也会调用上面的fulfillConnectPromise，进而触发后面的channelActive  
106.                 unsafe.finishConnect();  
107.             \}

上面的连接过程也是比较简单：

1、发起连接请求。1.1如果立即成功则执行连接成功的后续处理，如channelActive方法的调用；1.2如果未立即成功，则将连接事件注册到Selector；

2、Selector检测到连接事件后触发unsafe.finishConnect，该方法最终也执行连接成功的后续处理（同1.1）;

3、1.2的分支会创建一个超时检测任务，如果超过指定时间未连接成功，则直接关闭此次连接请求。  
![Center][]

[view plain]: https://blog.csdn.net/youaremoon/article/details/50538666#
[Center]: /images/20220527/110feccbcfc04fb8b56bbd57405ab4b6.png