Netty(七)——客户端创建之意

痛定思痛。 2022-04-18 05:26 270阅读 0赞

上一篇，看了Netty创建Server的步骤，和一些源码跟踪，这篇来看看Client创建的流程以及一些源码分析，还是我们到前边博客中，随便找一个完整的Client创建的示例代码来看。

Netty创建Client的流程其实和Server的挺类似的，下边，还是来看一张时序图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70][]

下边，来梳理一下上边的创建流程：1，用户创建Bootstrap实例，通过API设置创建客户端相关的参数，异步发起客户端连接；2，创建客户端连接、I/O读写的Reactor线程组NioEventLoopGroup，可以通过构造函数指定I/O线程的个数，默认为CPU内核的2倍；3，通过Bootstrap的ChannelFactory和用户指定的Channel类型创建用于客户端连接的NioSocketChannel；4，创建默认的Channel Handler Pipeline用于调度和执行网络事件；5，异步发起TCP连接，判断连接是否成功，如果成功直接将NioSocketChannel注册到多路复用器上，监听读操作位，如果没有没连接成功，则注册连接监听位到多路复用器，等待连接结果；6，注册对应的网络监听状态位到多路复用器；7，由多路复用器在I/O中轮询各Channel，处理连接结果；8，如果连接成功，设置Future结果，发送连接成功事件，出发ChannelPipeline执行；9，由ChannelPipeline调度执行系统和用户的ChannelHandler，执行业务逻辑。

看完流程后，接下来，我们看看源码吧。Bootstrap是外观模式的外观类，我们先来看看它。

1，group方法，设置I/O线程组的接口，AbstractBootstrap类中：

/**
         * The {@link EventLoopGroup} which is used to handle all the events for the to-be-creates
         * {@link Channel}
         */
        @SuppressWarnings("unchecked")
        public B group(EventLoopGroup group) {
            if (group == null) {
                throw new NullPointerException("group");
            }
            if (this.group != null) {
                throw new IllegalStateException("group set already");
            }
            this.group = group;
            return (B) this;
        }

2，option方法，TCP参数的设置接口，AbstractBootstrap类中：

/**
 * Allow to specify a {@link ChannelOption} which is used for the {@link Channel} instances once they got
 * created. Use a value of {@code null} to remove a previous set {@link ChannelOption}.
 */
 @SuppressWarnings("unchecked")
 public <T> B option(ChannelOption<T> option, T value) {
 if (option == null) {
 throw new NullPointerException("option");
 }
 if (value == null) {
 synchronized (options) {
 options.remove(option);
 }
 } else {
 synchronized (options) {
 options.put(option, value);
 }
 }
 return (B) this;
 }

在，ChanelOption常量类中，还有其它Tcp的参数，这里看几个：

<table> 
 <tbody> 
 <tr> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:1.7534in;"> 值 </td> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:5.3472in;"> 内容 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:1.7534in;"> SO_TIMEOUT </td> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:5.4166in;"> 控制读取操作将阻塞多少毫秒，如果返回值为0，计时器被禁止，该线程将无限期阻塞，直至成功； </td> 
 </tr> 
 <tr> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:1.7534in;"> SO_SNDBUF </td> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:5.3472in;"> 套接字使用的发送缓冲区大小 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:1.7534in;"> SO_RCVBUF </td> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:5.3472in;"> 套接字使用的接收缓冲区大小 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:1.7534in;"> SO_REUSEADDR </td> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:5.4013in;"> 用于决定如果网络上任然有数据向旧的ServerSocket传输数据，是否允许新的ServerSocket绑定到与旧的一样； </td> 
 </tr> 
 <tr> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:1.7729in;"> CONNECT_TIMEOUT_MILLIS </td> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:5.3277in;"> 客户端呢连接超时间 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:1.7534in;"> TCP_NODELAY </td> 
 <td style="border-color:#a3a3a3;vertical-align:top;width:5.3472in;"> 激活或禁止TCP——NODELAY套接字选项，它决定是否使用Nagle算法。 </td> 
 </tr> 
 </tbody> 
</table>

3，channel方法：通过工厂反射创建NioSocketChannel：

/**
 * The {@link Class} which is used to create {@link Channel} instances from.
 * You either use this or {@link #channelFactory(io.netty.channel.ChannelFactory)} if your
 * {@link Channel} implementation has no no-args constructor.
 */
 public B channel(Class<? extends C> channelClass) {
 if (channelClass == null) {
 throw new NullPointerException("channelClass");
 }
 return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
 }

4，设置Handler接口，通过ChannelInitializer继承ChannleHandlerAdapter，调用它的initChannel来设置用户的ChannelHandler。类ChannelInitializer中：

public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            ChannelPipeline pipeline = ctx.pipeline();
            boolean success = false;
            try {
                initChannel((C) ctx.channel());
                pipeline.remove(this);
                ctx.fireChannelRegistered();
                success = true;
            } catch (Throwable t) {
                logger.warn("Failed to initialize a channel. Closing: " + ctx.channel(), t);
            } finally {
                if (pipeline.context(this) != null) {
                    pipeline.remove(this);
                }
                if (!success) {
                    ctx.close();
                }
            }
        }

设置完Bootstrap，下边来看看发起客户端连接吧connect吧。

1，首先创建和初始化NioSocketChannel，eventLoop：

/**
 * @see {@link #connect()}
 */
 private ChannelFuture doResolveAndConnect(SocketAddress remoteAddress, final SocketAddress localAddress) {
 //1，初始化ChannelFuture
 final ChannelFuture regFuture = initAndRegister();
 if (regFuture.cause() != null) {
 return regFuture;
 }
 //2,channel得到
 final Channel channel = regFuture.channel();
 //3,eventLoop线程得到
 final EventLoop eventLoop = channel.eventLoop();
 final NameResolver<SocketAddress> resolver = this.resolver.getResolver(eventLoop);
 
 if (!resolver.isSupported(remoteAddress) || resolver.isResolved(remoteAddress)) {
 // Resolver has no idea about what to do with the specified remote address or it's resolved already.
 return doConnect(remoteAddress, localAddress, regFuture, channel.newPromise());
 }
 
 final Future<SocketAddress> resolveFuture = resolver.resolve(remoteAddress);
 final Throwable resolveFailureCause = resolveFuture.cause();
 
 if (resolveFailureCause != null) {
 // Failed to resolve immediately
 channel.close();
 return channel.newFailedFuture(resolveFailureCause);
 }
 
 if (resolveFuture.isDone()) {
 // Succeeded to resolve immediately; cached? (or did a blocking lookup)
 return doConnect(resolveFuture.getNow(), localAddress, regFuture, channel.newPromise());
 }
 
 // Wait until the name resolution is finished.
 final ChannelPromise connectPromise = channel.newPromise();
 resolveFuture.addListener(new FutureListener<SocketAddress>() {
 @Override
 public void operationComplete(Future<SocketAddress> future) throws Exception {
 if (future.cause() != null) {
 channel.close();
 connectPromise.setFailure(future.cause());
 } else {
 doConnect(future.getNow(), localAddress, regFuture, connectPromise);
 }
 }
 });
 
 return connectPromise;
 }

2，看下发起异步TCP连接吧，doConnect里边进行找吧：

private static void doConnect0(
                final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelFuture regFuture,
                final ChannelPromise connectPromise) {
    
            // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
            // the pipeline in its channelRegistered() implementation.
            final Channel channel = connectPromise.channel();
            channel.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    if (regFuture.isSuccess()) {
                        if (localAddress == null) {
                            channel.connect(remoteAddress, connectPromise);
                        } else {
                            channel.connect(remoteAddress, localAddress, connectPromise);
                        }
                        connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                    } else {
                        connectPromise.setFailure(regFuture.cause());
                    }
                }
            });
        }

最终调用的是channel的connect方法，这里看下其中代码，包括具体调用NioSocketChannel的连接动作，连接成功，失败的各种处理，还有就是定时对连接超时的处理等：

@Override
            public final void connect(
                    final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
                if (!promise.setUncancellable() || !ensureOpen(promise)) {
                    return;
                }
    
                try {
                    if (connectPromise != null) {
                        throw new IllegalStateException("connection attempt already made");
                    }
    
                    boolean wasActive = isActive();
                    //连接动作，可以具体再看NioSocketChannel类中的连接处理
                    if (doConnect(remoteAddress, localAddress)) {
                        //连接成功后，触发链路激活事件，在pipeline进行传播
                        fulfillConnectPromise(promise, wasActive);
                    } else {
                        connectPromise = promise;
                        requestedRemoteAddress = remoteAddress;
                        
                        //根据设置的连接超时时间，启动连接超时检测定时器
                        // Schedule connect timeout.
                        int connectTimeoutMillis = config().getConnectTimeoutMillis();
                        if (connectTimeoutMillis > 0) {
                            connectTimeoutFuture = eventLoop().schedule(new OneTimeTask() {
                                @Override
                                public void run() {
                                    ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                                    ConnectTimeoutException cause =
                                            new ConnectTimeoutException("connection timed out: " + remoteAddress);
                                    if (connectPromise != null && connectPromise.tryFailure(cause)) {
                                        close(voidPromise());
                                    }
                                }
                            }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
                        }
    
                        promise.addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture future) throws Exception {
                                if (future.isCancelled()) {
                                    if (connectTimeoutFuture != null) {
                                        connectTimeoutFuture.cancel(false);
                                    }
                                    connectPromise = null;
                                    close(voidPromise());
                                }
                            }
                        });
                    }
                } catch (Throwable t) {
                    promise.tryFailure(annotateConnectException(t, remoteAddress));
                    closeIfClosed();
                }
            }
    
    
    
    private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
                if (promise == null) {
                    // Closed via cancellation and the promise has been notified already.
                    return;
                }
    
                // trySuccess() will return false if a user cancelled the connection attempt.
                boolean promiseSet = promise.trySuccess();
    
                // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
                // because what happened is what happened.
                if (!wasActive && isActive()) {
                    pipeline().fireChannelActive();
                }
    
                // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
                if (!promiseSet) {
                    close(voidPromise());
                }
            }

好，这篇大概讲了Netty客户端创建的流程，和流程源码的查看，其中还有很多细节，需要去查看，从中可以学习作者的一些思想，一些设计思路，还是非常好的，另外多看看英文注解，写的还是挺全的。  继续学习中吧。。。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70]: /images/20220418/4bacb8dda95b4be4b38311a8f209d0f2.png