源码分析Dubbo网络通信NettyClient实现原理

丁威

发布于 2019-06-10 17:28:59

1.1K0

发布于 2019-06-10 17:28:59

文章被收录于专栏：中间件兴趣圈中间件兴趣圈中间件兴趣圈

微信公众号：[中间件兴趣圈] 作者简介：《RocketMQ技术内幕》作者

在源码分析Dubbo通讯篇之网络核心类一文中已给出Dubbo netty client的启动流程，如下图：

以Dubbo协议为例，DubboProtocol#refer中，在创建Invoker时，通过getClient方法，开始Client（连接的）创建过程，先重点看一下：

 1private ExchangeClient[] getClients(URL url) {    // @1
 2        // whether to share connection
 3        boolean service_share_connect = false;
 4        int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);    // @2
 5        // if not configured, connection is shared, otherwise, one connection for one service      
 6        if (connections == 0) {
 7            service_share_connect = true;
 8            connections = 1;
 9        }
10
11        ExchangeClient[] clients = new ExchangeClient[connections];   // @3
12        for (int i = 0; i < clients.length; i++) {
13            if (service_share_connect) {
14                clients[i] = getSharedClient(url);       // @4
15            } else {
16                clients[i] = initClient(url);    // @5
17            }
18        }
19        return clients;
20    }

代码@1：参数URL,服务提供者URL。

代码@2：获取<dubbo:reference connections = "" />，默认0表示客户端对同一个服务提供者的所有服务，使用共享一个连接，如果该值有设置，则使用非共享的客户端，所谓的共享客户端，以Netty为例，也即客户端对同一服务提供者发起的不同服务，使用同一个客户端(NettyClient)进行请求的发送与接收。

代码@3：根据connections，创建ExchangeClients数组。

代码@4：如果使用共享连接，则调用getSharedClient获取共享连接，如果客户端未建立，则创建客户端。

代码@5：如果不使用共享连接，调用initClient创建客户端，其创建时序图如上图所示。

接下来，还是以Netty4为例，探究一下Dubbo NettyClient的创建细节。

首先从全貌上大概看一下NettyClient对象所持有的属性：

AbstractPeer 1、private final ChannelHandler handler 事件处理Handler。 2、private volatile URL url 该协议的第一个服务提供者的URL，Server只需要用到URL中的参数，与具体某一个服务没什么关系。
AbstractEndpoint 1、private Codec2 codec 编码解码器。 2、private int timeout 超时时间 3、private int connectTimeout 连接超时时间注，如果通过dubbo:provider改变codec,不同的服务提供者引用的默认服务提供者参数不同，那这个只能是以第一个为主了，应该不科学？
AbstractClient 1、protected static final String CLIENT_THREAD_POOL_NAME = "DubboClientHandler" Dubbo 客户端线程名称。 2、private static final AtomicInteger CLIENT_THREAD_POOL_ID = new AtomicInteger() 客户端线程池ID自增器。 3、private static final ScheduledThreadPoolExecutor reconnectExecutorService 客户端连接重连线程池。 4、private final Lock connectLock = new ReentrantLock() 客户端连接服务端独占锁，保证一个客户端同时只会一个线程在执行连接动作。 5、private final boolean send_reconnect 消息发送时，如果当前客户端未连接，是否发起重连操作。 6、private final AtomicInteger reconnect_count = new AtomicInteger(0) 记录重连的次数。 7、private final AtomicBoolean reconnect_error_log_flag = new AtomicBoolean(false) 连接出错后是否打印过ERROR日志。 8、private final int reconnect_warning_period 对连接异常，以WARN级别日志输出的频率，默认第一次是以Error日志，然后每出现reconnect_warning_period次后，就打印一次warn级别日志。 9、private final long shutdown_timeout 关闭服务的超时时间。 10、protected volatile ExecutorService executor 客户端线程池。 11、private volatile ScheduledFuture< ?> reconnectExecutorFuture = null 重连的Future。 12、private long lastConnectedTime = System.currentTimeMillis() 上一次重连时间戳。
NettyClient 1、private static final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(Constants.DEFAULT_IO_THREADS, new DefaultThreadFactory("NettyClientWorker", true)) IO线程组，同一个JVM中所有的客户端公用一个IO线程组，且线程数固定为(32与CPU核数+1的最小值)。 2、private Bootstrap bootstrap Netty客户端启动实例。 3、private volatile Channel channel 客户端连接，请copy其引用使用。

源码分析NettyClient

构造函数

1public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException {    // @1
2        super(url, wrapChannelHandler(url, handler));    // @2
3}

代码@1：url:服务提供者URL；ChannelHandler handler：事件处理器。

代码@2：wrapChannelHandler在讲解NettyServer时已重点分析，构造其事件转发模型(Dispatch)。

接下来重点分析其父类的构造方法：

 1public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
 2        super(url, handler);     // @1
 3
 4        send_reconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, false);
 5
 6        shutdown_timeout = url.getParameter(Constants.SHUTDOWN_TIMEOUT_KEY, Constants.DEFAULT_SHUTDOWN_TIMEOUT);
 7
 8        // The default reconnection interval is 2s, 1800 means warning interval is 1 hour.
 9        reconnect_warning_period = url.getParameter("reconnect.waring.period", 1800);    // @2
10
11        try {
12            doOpen();        // @3
13        } catch (Throwable t) {
14            close();
15            throw new RemotingException(url.toInetSocketAddress(), null,
16                    "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
17                            + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
18        }
19        try {
20            // connect.
21            connect();      // @4
22            if (logger.isInfoEnabled()) {
23                logger.info("Start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress());
24            }
25        } catch (RemotingException t) {
26            if (url.getParameter(Constants.CHECK_KEY, true)) {
27                close();
28                throw t;
29            } else {
30                logger.warn("Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
31                        + " connect to the server " + getRemoteAddress() + " (check == false, ignore and retry later!), cause: " + t.getMessage(), t);
32            }
33        } catch (Throwable t) {
34            close();
35            throw new RemotingException(url.toInetSocketAddress(), null,
36                    "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
37                            + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
38        }
39
40        executor = (ExecutorService) ExtensionLoader.getExtensionLoader(DataStore.class)    // @5
41                .getDefaultExtension().get(Constants.CONSUMER_SIDE, Integer.toString(url.getPort()));
42        ExtensionLoader.getExtensionLoader(DataStore.class)
43                .getDefaultExtension().remove(Constants.CONSUMER_SIDE, Integer.toString(url.getPort()));
44    }

代码@1：调用父类的构造其，初始化url、ChannelHandler。

代码@2：初始化send_reconnect、shutdown_timeout、reconnect_warning_period（默认1小时打印一次日志）

代码@3：调用doOpen初始化客户端调用模型，后续重点分析。

代码@4：调用connect方法，向服务端发起TCP连接。

代码@5：获取线程池，并从缓存中移除。

doOpen

 1protected void doOpen() throws Throwable {
 2        NettyHelper.setNettyLoggerFactory();
 3        final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this);    // @1 
 4        bootstrap = new Bootstrap();                                                                                            // @2
 5        bootstrap.group(nioEventLoopGroup)                                                                              // @3
 6                .option(ChannelOption.SO_KEEPALIVE, true)
 7                .option(ChannelOption.TCP_NODELAY, true)
 8                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
 9                //.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())
10                .channel(NioSocketChannel.class);
11
12        if (getTimeout() < 3000) {     // @4
13            bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
14        } else {
15            bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout());
16        }
17
18        bootstrap.handler(new ChannelInitializer() {
19
20            @Override
21            protected void initChannel(Channel ch) throws Exception {
22                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
23                ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
24                        .addLast("decoder", adapter.getDecoder())
25                        .addLast("encoder", adapter.getEncoder())
26                        .addLast("handler", nettyClientHandler);      // @5
27            }
28        });
29    }

代码@1：创建NettyClientHandler。

代码@2：创建Netty客户端启动实例bootstrap.

代码@3：客户端绑定IO线程组（池），注意，一个JVM中所有的NettyClient共享其IO线程。

代码@4：设置连接超时时间，最小连接超时时间为3s。

代码@5：设置编码器、事件连接器，当触发事件后，将调用nettyClientHandler中相关的方法。

doConnect

 1protected void doConnect() throws Throwable {
 2        long start = System.currentTimeMillis();
 3        ChannelFuture future = bootstrap.connect(getConnectAddress());   // @1
 4        try {
 5            boolean ret = future.awaitUninterruptibly(3000, TimeUnit.MILLISECONDS);   // @2
 6
 7            if (ret && future.isSuccess()) {
 8                Channel newChannel = future.channel();
 9                try {
10                    // Close old channel
11                    Channel oldChannel = NettyClient.this.channel; // copy reference
12                    if (oldChannel != null) {
13                        try {
14                            if (logger.isInfoEnabled()) {
15                                logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
16                            }
17                            oldChannel.close();
18                        } finally {
19                            NettyChannel.removeChannelIfDisconnected(oldChannel);
20                        }
21                    }
22                } finally {
23                    if (NettyClient.this.isClosed()) {
24                        try {
25                            if (logger.isInfoEnabled()) {
26                                logger.info("Close new netty channel " + newChannel + ", because the client closed.");
27                            }
28                            newChannel.close();
29                        } finally {
30                            NettyClient.this.channel = null;
31                            NettyChannel.removeChannelIfDisconnected(newChannel);
32                        }
33                    } else {
34                        NettyClient.this.channel = newChannel;
35                    }
36                }
37            } else if (future.cause() != null) {
38                throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
39                        + getRemoteAddress() + ", error message is:" + future.cause().getMessage(), future.cause());
40            } else {
41                throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
42                        + getRemoteAddress() + " client-side timeout "
43                        + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client "
44                        + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion());
45            }
46        } finally {
47            if (!isConnected()) {
48                //future.cancel(true);
49            }
50        }
51    }

代码@1：调用bootstrap.connect方法发起TCP连接。

代码@2：future.awaitUninterruptibly，连接事件只等待3S，这里写成固定了，显然没有与doOpen方法中ChannelOption.CONNECT_TIMEOUT_MILLIS保持一致。

关于NettyClient的介绍就将到这里了，下一篇将会分析编码解码。

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2019-03-07，如有侵权请联系 cloudcommunity@tencent.com 删除

dubbo

本文分享自中间件兴趣圈微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

dubbo

登录后参与评论

0 条评论

热度