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

83 阅读 0 评论 55 点赞

我是靠谱客的博主俏皮发卡，最近开发中收集的这篇文章主要介绍Netty(七)——客户端创建之意，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

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

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

下边，来梳理一下上边的创建流程：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的参数，这里看几个：

值	内容
SO_TIMEOUT	控制读取操作将阻塞多少毫秒，如果返回值为0，计时器被禁止，该线程将无限期阻塞，直至成功；
SO_SNDBUF	套接字使用的发送缓冲区大小
SO_RCVBUF	套接字使用的接收缓冲区大小
SO_REUSEADDR	用于决定如果网络上任然有数据向旧的ServerSocket传输数据，是否允许新的ServerSocket绑定到与旧的一样；
CONNECT_TIMEOUT_MILLIS	客户端呢连接超时间
TCP_NODELAY	激活或禁止TCP——NODELAY套接字选项，它决定是否使用Nagle算法。

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客户端创建的流程，和流程源码的查看，其中还有很多细节，需要去查看，从中可以学习作者的一些思想，一些设计思路，还是非常好的，另外多看看英文注解，写的还是挺全的。继续学习中吧。。。