Dubbo分析之Transport层
系列文章
Dubbo分析Serialize层
Dubbo分析之Transport层
Dubbo分析之Exchange 层
Dubbo分析之Protocol层
前言
上一篇文章Dubbo分析之Serialize层,介绍了最底层的序列化/反序列化层,本文继续分析Serialize层的上一层transport网络传输层,此层使用了现有的一些通讯开源框架(ex:netty,mina,grizzly)来做底层通讯,上文也做了简单介绍,本文将做更深入的了解;
Transporter类分析
dubbo为通讯框架提供了统一的bind和connet接口,方便进行管理和扩展,封装在接口类:Transporter中:
@SPI("netty") public interface Transporter { @Adaptive({Constants.SERVER_KEY, Constants.TRANSPORTER_KEY}) Server bind(URL url, ChannelHandler handler) throws RemotingException; @Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY}) Client connect(URL url, ChannelHandler handler) throws RemotingException; }
提供了bind和connect接口,分别对应这服务器端和客户端,具体有哪些实现类,如下图所示:
以默认使用的netty框架为例,代码如下:
public class NettyTransporter implements Transporter { public static final String NAME = "netty"; @Override public Server bind(URL url, ChannelHandler listener) throws RemotingException { return new NettyServer(url, listener); } @Override public Client connect(URL url, ChannelHandler listener) throws RemotingException { return new NettyClient(url, listener); } }
具体的服务器端封装在NettyServer中,客户端封装在NettyClient;url参数是包含了xml配置的信息(包括:对外的接口,使用的协议,使用的序列化方式,使用的通讯框架等),listener是一个Handler,在解码之后将数据交给它做后续的业务处理;对应以上的几种通讯开源框架,分别提供了对应的Transporter包括:NettyTransporter,NettyTransporter(netty4),MinaTransporter以及GrizzlyTransporter,具体使用哪种类型的Transporter,在Transporters类中提供了getTransporter方法:
public static Transporter getTransporter() { return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension(); }
这里并没有像在获取具体serialization类一样,通过在url指定transporter参数,然后加载具体的transporter类,而是生成了一个动态的transporter,由此动态transporter去加载具体的类;
因为Server端和Client可以分别设置成不同的通讯框架,一次获取唯一的Transporter不能满足此需求;具体的生成动态代码的方法在ExtensionLoader的createAdaptiveExtensionClassCode方法中,此处不在列出源码,在此展示一下默认生成的动态代码扩展类:
package com.alibaba.dubbo.remoting; import com.alibaba.dubbo.common.extension.ExtensionLoader; public class Transporter$Adaptive implements com.alibaba.dubbo.remoting.Transporter { public com.alibaba.dubbo.remoting.Server bind( com.alibaba.dubbo.common.URL arg0, com.alibaba.dubbo.remoting.ChannelHandler arg1) throws com.alibaba.dubbo.remoting.RemotingException { if (arg0 == null) { throw new IllegalArgumentException("url == null"); } com.alibaba.dubbo.common.URL url = arg0; String extName = url.getParameter("server", url.getParameter("transporter", "netty")); if (extName == null) { throw new IllegalStateException( "Fail to get extension(com.alibaba.dubbo.remoting.Transporter) name from url(" + url.toString() + ") use keys([server, transporter])"); } com.alibaba.dubbo.remoting.Transporter extension = (com.alibaba.dubbo.remoting.Transporter) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.remoting.Transporter.class) .getExtension(extName); return extension.bind(arg0, arg1); } public com.alibaba.dubbo.remoting.Client connect( com.alibaba.dubbo.common.URL arg0, com.alibaba.dubbo.remoting.ChannelHandler arg1) throws com.alibaba.dubbo.remoting.RemotingException { if (arg0 == null) { throw new IllegalArgumentException("url == null"); } com.alibaba.dubbo.common.URL url = arg0; String extName = url.getParameter("client", url.getParameter("transporter", "netty")); if (extName == null) { throw new IllegalStateException( "Fail to get extension(com.alibaba.dubbo.remoting.Transporter) name from url(" + url.toString() + ") use keys([client, transporter])"); } com.alibaba.dubbo.remoting.Transporter extension = (com.alibaba.dubbo.remoting.Transporter) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.remoting.Transporter.class) .getExtension(extName); return extension.connect(arg0, arg1); } }
可以发现Server端可以通过transporter和server两个参数来设置扩展类,而且server参数设置的值是可以覆盖transporter参数的值,同理Client也类似;最后不管是bind()还是connet()都是通过ExtensionLoader的getExtension方法来获取具体的transporter类;同serialize层,相关的transporter也同样定义在META-INF/dubbo/internal/com.alibaba.dubbo.remoting.Transporter文件中:
netty=com.alibaba.dubbo.remoting.transport.netty.NettyTransporter netty4=com.alibaba.dubbo.remoting.transport.netty4.NettyTransporter mina=com.alibaba.dubbo.remoting.transport.mina.MinaTransporter grizzly=com.alibaba.dubbo.remoting.transport.grizzly.GrizzlyTransporter
Server端和Client分析
1.Server端
在实例化具体的Server类时,会首先调用父类的构造器,进行参数初始化,同时调用bind()方法,启动服务器;父类AbstractServer构造器如下:
public AbstractServer(URL url, ChannelHandler handler) throws RemotingException { super(url, handler); localAddress = getUrl().toInetSocketAddress(); String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost()); int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort()); if (url.getParameter(Constants.ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) { bindIp = NetUtils.ANYHOST; } bindAddress = new InetSocketAddress(bindIp, bindPort); this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS); this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT); try { doOpen(); if (logger.isInfoEnabled()) { logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress()); } } catch (Throwable t) { throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName() + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t); } //fixme replace this with better method DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension(); executor = (ExecutorService) dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY, Integer.toString(url.getPort())); }
主要从url获取启动参数包括:ip,port,accepts(可接受的连接数,0表示不受限制数量,默认为0),idleTimeout等;然后调用doOpen方法通过具体的通讯框架绑定端口启动服务;已默认使用的Netty为例,查看doOpen()方法如下:
protected void doOpen() throws Throwable { NettyHelper.setNettyLoggerFactory(); ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true)); ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true)); ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS)); bootstrap = new ServerBootstrap(channelFactory); final NettyHandler nettyHandler = new NettyHandler(getUrl(), this); channels = nettyHandler.getChannels(); // https://issues.jboss.org/browse/NETTY-365 // https://issues.jboss.org/browse/NETTY-379 // final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true)); bootstrap.setOption("child.tcpNoDelay", true); bootstrap.setPipelineFactory(new ChannelPipelineFactory() { @Override public ChannelPipeline getPipeline() { NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this); ChannelPipeline pipeline = Channels.pipeline(); /*int idleTimeout = getIdleTimeout(); if (idleTimeout > 10000) { pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0)); }*/ pipeline.addLast("decoder", adapter.getDecoder()); pipeline.addLast("encoder", adapter.getEncoder()); pipeline.addLast("handler", nettyHandler); return pipeline; } }); // bind channel = bootstrap.bind(getBindAddress()); }
以上是常规的启动netty程序,需要指定编解码器,nettyHandler;编解码已经在上文中介绍过了,此处不在详细介绍,重点介绍nettyHandler;server端在数据经过解码之后就交给NettyHandler来处理,NettyHandler继承于Netty的SimpleChannelHandler类,重写了channelConnected,channelDisconnected,messageReceived,writeRequested以及exceptionCaught方法,基本上就是常规的几种操作:建立连接,断开连接,接收消息,发送消息,异常处理;看一下部分源码:
@Override public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler); try { if (channel != null) { channels.put(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()), channel); } handler.connected(channel); } finally { NettyChannel.removeChannelIfDisconnected(ctx.getChannel()); } } @Override public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler); try { channels.remove(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress())); handler.disconnected(channel); } finally { NettyChannel.removeChannelIfDisconnected(ctx.getChannel()); } } @Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler); try { handler.received(channel, e.getMessage()); } finally { NettyChannel.removeChannelIfDisconnected(ctx.getChannel()); } } @Override public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception { super.writeRequested(ctx, e); NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler); try { handler.sent(channel, e.getMessage()); } finally { NettyChannel.removeChannelIfDisconnected(ctx.getChannel()); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler); try { handler.caught(channel, e.getCause()); } finally { NettyChannel.removeChannelIfDisconnected(ctx.getChannel()); } }
将netty原生的channel包装成dubbo的NettyChannel,同时将NettyChannel保存在NettyChannel的内部静态变量channelMap中;这里的方法都统一调用了getOrAddChannel方法,先添加进去,最后在finally中判定channel是否已经关闭,如果关闭从channelMap中移除;中间部分调用了handler对应的方法,此处的handler就是在实例化时传入的NettyServer,NettyServer本身也是一个ChannelHandler,可以看一下channelHandler接口类:
public interface ChannelHandler { void connected(Channel channel) throws RemotingException; void disconnected(Channel channel) throws RemotingException; void sent(Channel channel, Object message) throws RemotingException; void received(Channel channel, Object message) throws RemotingException; void caught(Channel channel, Throwable exception) throws RemotingException; }
具体的server类中也可以做一些处理,比如connected时判段是否超过accepts,如果超过拒绝连接;处理完之后交给实例化Server时传入的ChannelHandler处理,此类具体是在HeaderExchanger中被初始化的:
public class HeaderExchanger implements Exchanger { public static final String NAME = "header"; @Override public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true); } @Override public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))); } }
可以发现这里具体的ChannelHandler是DecodeHandler,注这里的Decode和Netty本身的decode不一样,Netty本身的decode在执行NettyHandler之前就执行解码了;后续的操作在Exchange层进行处理,本文暂时先不做介绍;
2.Client端
同样查看父类AbstractClient,构造方法如下:
public AbstractClient(URL url, ChannelHandler handler) throws RemotingException { super(url, handler); send_reconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, false); shutdown_timeout = url.getParameter(Constants.SHUTDOWN_TIMEOUT_KEY, Constants.DEFAULT_SHUTDOWN_TIMEOUT); // The default reconnection interval is 2s, 1800 means warning interval is 1 hour. reconnect_warning_period = url.getParameter("reconnect.waring.period", 1800); try { doOpen(); } catch (Throwable t) { close(); throw new RemotingException(url.toInetSocketAddress(), null, "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t); } try { // connect. connect(); if (logger.isInfoEnabled()) { logger.info("Start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress()); } } catch (RemotingException t) { if (url.getParameter(Constants.CHECK_KEY, true)) { close(); throw t; } else { logger.warn("Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress() + " (check == false, ignore and retry later!), cause: " + t.getMessage(), t); } } catch (Throwable t) { close(); throw new RemotingException(url.toInetSocketAddress(), null, "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t); } executor = (ExecutorService) ExtensionLoader.getExtensionLoader(DataStore.class) .getDefaultExtension().get(Constants.CONSUMER_SIDE, Integer.toString(url.getPort())); ExtensionLoader.getExtensionLoader(DataStore.class) .getDefaultExtension().remove(Constants.CONSUMER_SIDE, Integer.toString(url.getPort())); }
客户端需要提供重连机制,所以初始化的几个参数都和重连有关,send_reconnect表示在发送消息时发现连接已经断开是否发起重连,reconnect_warning_period表示多久报一次重连警告,shutdown_timeout表示连接服务器一直连接不上的超时时间;接下来就是调用doOpen()方法,同样已Netty为例:
protected void doOpen() throws Throwable { NettyHelper.setNettyLoggerFactory(); bootstrap = new ClientBootstrap(channelFactory); // config // @see org.jboss.netty.channel.socket.SocketChannelConfig bootstrap.setOption("keepAlive", true); bootstrap.setOption("tcpNoDelay", true); bootstrap.setOption("connectTimeoutMillis", getTimeout()); final NettyHandler nettyHandler = new NettyHandler(getUrl(), this); bootstrap.setPipelineFactory(new ChannelPipelineFactory() { @Override public ChannelPipeline getPipeline() { NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this); ChannelPipeline pipeline = Channels.pipeline(); pipeline.addLast("decoder", adapter.getDecoder()); pipeline.addLast("encoder", adapter.getEncoder()); pipeline.addLast("handler", nettyHandler); return pipeline; } }); }
Netty客户端的常规代码,设置了和Server端相同的NettyHandler,decoder和encoder;下面重点看看connect方法:
protected void connect() throws RemotingException { connectLock.lock(); try { if (isConnected()) { return; } initConnectStatusCheckCommand(); doConnect(); if (!isConnected()) { throw new RemotingException(this, "Failed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " " + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion() + ", cause: Connect wait timeout: " + getTimeout() + "ms."); } else { if (logger.isInfoEnabled()) { logger.info("Successed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " " + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion() + ", channel is " + this.getChannel()); } } reconnect_count.set(0); reconnect_error_log_flag.set(false); } catch (RemotingException e) { throw e; } catch (Throwable e) { throw new RemotingException(this, "Failed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " " + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion() + ", cause: " + e.getMessage(), e); } finally { connectLock.unlock(); } }
首先判定是否已经连接,如果连接直接return;接下来初始化连接状态检查器,定期检查channel是否连接,连接断开会进行重连操作,具体代码如下:
private synchronized void initConnectStatusCheckCommand() { //reconnect=false to close reconnect int reconnect = getReconnectParam(getUrl()); if (reconnect > 0 && (reconnectExecutorFuture == null || reconnectExecutorFuture.isCancelled())) { Runnable connectStatusCheckCommand = new Runnable() { @Override public void run() { try { if (!isConnected()) { connect(); } else { lastConnectedTime = System.currentTimeMillis(); } } catch (Throwable t) { String errorMsg = "client reconnect to " + getUrl().getAddress() + " find error . url: " + getUrl(); // wait registry sync provider list if (System.currentTimeMillis() - lastConnectedTime > shutdown_timeout) { if (!reconnect_error_log_flag.get()) { reconnect_error_log_flag.set(true); logger.error(errorMsg, t); return; } } if (reconnect_count.getAndIncrement() % reconnect_warning_period == 0) { logger.warn(errorMsg, t); } } } }; reconnectExecutorFuture = reconnectExecutorService.scheduleWithFixedDelay(connectStatusCheckCommand, reconnect, reconnect, TimeUnit.MILLISECONDS); } }
创建了一个Runnable,用来检测是否连接,如果连接断开,调用connect方法;定时调度交给ScheduledThreadPoolExecutor来执行;初始化之后就调用具体Client的doConnect操作,也是通讯框架的一些常规代码,此处不列出了;后续关于NettyChannel的介绍和Server端类似,不过多进行介绍;
总结
本文重点分析了dubbo架构中的transport层,具体围绕Transporter, Client, Server,ChannelHandler几个类展开,关于后续的处理将在exchange信息交换层;
示例代码地址
https://github.com/ksfzhaohui...
https://gitee.com/OutOfMemory...
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
在Vue项目里面使用d3.js
之前写一个 Demo里面 有些东西要使用d3实现一些效果 但是在很多论坛找资源都找不到可以在Vue里面使用D3.js的方法,npm 上面的D3相对来说 可以说是很不人性化了 完全没有说 在webpack上怎么使用D3.js 最后折腾很久 看到某位外国大佬 看他的案例 成功的实现了在Vue项目里面实现D3的使用 首先安装 npm install d3 --save-dev 以防万一,然后看package.json 安装完成 在我们开始之前,让我们渲染一个Vue组件,它使用常规的D3 DOM操作呈现一个简单的折线图: <script> import * as d3 from 'd3'; const data = [99, 71, 78, 25, 36, 92]; export default { name: 'non-vue-line-chart', template: '<div></div>', mounted() { const svg = d3.select(this.$el) .append('svg') .attr('width', 500)...
- 下一篇
安装kubernetes1.12.1的 dashboard v1.10 + Heapster
Dashboard是kubernetes的官方WEB UI。 Heapster为集群添加使用统计和监控功能,为Dashboard添加仪表盘。 使用InfluxDB做为Heapster的后端存储。 Dashboard 安装 kubernetes dashboard 官方资源定义文档:https://raw.githubusercontent.com/kubernetes/dashboard/master/src/deploy/recommended/kubernetes-dashboard.yaml 注意点: 默认资源定义文档中Service 定义没有使用NodePort,不能服务器外部访问 默认资源定义文档中的权限定义,仅包含了dashboard需要的最小权限,不支持本地访问外的其他方式访问,需要创建身份令牌(Create An Authentication Token)才能独立的提供访问。 通过查看dashboard的定义文档,需要的镜像是k8s.gcr.io/kubernetes-dashboard-amd64:v1.10.0, 我们在所有node节点上pull该镜像: docke...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- SpringBoot2整合Thymeleaf,官方推荐html解决方案
- Windows10,CentOS7,CentOS8安装MongoDB4.0.16
- CentOS6,CentOS7官方镜像安装Oracle11G
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
- SpringBoot2更换Tomcat为Jetty,小型站点的福音
- CentOS7安装Docker,走上虚拟化容器引擎之路
- CentOS7编译安装Gcc9.2.0,解决mysql等软件编译问题
- Linux系统CentOS6、CentOS7手动修改IP地址