Transporter類分析
dubbo為通訊框架提供了統(tǒng)一的bind和connet接口犁钟，方便進(jìn)行管理和擴(kuò)展脓规，封裝在接口類：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接口士败，分別對應(yīng)這服務(wù)器端和客戶端送讲，具體有哪些實(shí)現(xiàn)類瞻想，如下圖所示：
以默認(rèn)使用的netty框架為例砾省，代碼如下：
?public class NettyTransporter implements Transporter {
? ? public static final String NAME = "netty";
public Server bind(URL url, ChannelHandler listener) throws RemotingException {
return new NettyServer(url, listener);
}
public Client connect(URL url, ChannelHandler listener) throws RemotingException {
return new NettyClient(url, listener);
}
}
具體的服務(wù)器端封裝在NettyServer中把跨，客戶端封裝在NettyClient谁鳍；url參數(shù)是包含了xml配置的信息(包括：對外的接口们拙，使用的協(xié)議稍途，使用的序列化方式，使用的通訊框架等)砚婆，listener是一個Handler械拍，在解碼之后將數(shù)據(jù)交給它做后續(xù)的業(yè)務(wù)處理；對應(yīng)以上的幾種通訊開源框架装盯，分別提供了對應(yīng)的Transporter包括：NettyTransporter坷虑，NettyTransporter(netty4)，MinaTransporter以及GrizzlyTransporter埂奈，具體使用哪種類型的Transporter迄损，在Transporters類中提供了getTransporter方法：
public static Transporter getTransporter() {
return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
}
這里并沒有像在獲取具體serialization類一樣，通過在url指定transporter參數(shù)账磺，然后加載具體的transporter類芹敌，而是生成了一個動態(tài)的transporter，由此動態(tài)transporter去加載具體的類垮抗；
因?yàn)镾erver端和Client可以分別設(shè)置成不同的通訊框架氏捞，一次獲取唯一的Transporter不能滿足此需求；具體的生成動態(tài)代碼的方法在ExtensionLoader的createAdaptiveExtensionClassCode方法中冒版，此處不在列出源碼液茎，在此展示一下默認(rèn)生成的動態(tài)代碼擴(kuò)展類：
?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);
}
}
可以發(fā)現(xiàn)Server端可以通過transporter和server兩個參數(shù)來設(shè)置擴(kuò)展類，而且server參數(shù)設(shè)置的值是可以覆蓋transporter參數(shù)的值辞嗡，同理Client也類似捆等；最后不管是bind()還是connet()都是通過ExtensionLoader的getExtension方法來獲取具體的transporter類；同serialize層欲间，相關(guān)的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端
在實(shí)例化具體的Server類時楚里，會首先調(diào)用父類的構(gòu)造器，進(jìn)行參數(shù)初始化猎贴，同時調(diào)用bind()方法，啟動服務(wù)器蝴光；父類AbstractServer構(gòu)造器如下：
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獲取啟動參數(shù)包括：ip她渴，port，accepts(可接受的連接數(shù)蔑祟，0表示不受限制數(shù)量趁耗，默認(rèn)為0)，idleTimeout等疆虚；然后調(diào)用doOpen方法通過具體的通訊框架綁定端口啟動服務(wù)苛败；已默認(rèn)使用的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());
}
以上是常規(guī)的啟動netty程序，需要指定編解碼器罢屈，nettyHandler嘀韧；編解碼已經(jīng)在上文中介紹過了，此處不在詳細(xì)介紹缠捌，重點(diǎn)介紹nettyHandler锄贷；server端在數(shù)據(jù)經(jīng)過解碼之后就交給NettyHandler來處理，NettyHandler繼承于Netty的SimpleChannelHandler類曼月，重寫了channelConnected谊却，channelDisconnected，messageReceived哑芹，writeRequested以及exceptionCaught方法炎辨，基本上就是常規(guī)的幾種操作：建立連接，斷開連接聪姿，接收消息碴萧，發(fā)送消息，異常處理咳燕；看一下部分源碼：
@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的內(nèi)部靜態(tài)變量channelMap中；這里的方法都統(tǒng)一調(diào)用了getOrAddChannel方法招盲，先添加進(jìn)去低缩，最后在finally中判定channel是否已經(jīng)關(guān)閉，如果關(guān)閉從channelMap中移除曹货；中間部分調(diào)用了handler對應(yīng)的方法咆繁，此處的handler就是在實(shí)例化時傳入的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，如果超過拒絕連接礼饱；處理完之后交給實(shí)例化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))));
}
}
可以發(fā)現(xiàn)這里具體的ChannelHandler是DecodeHandler，注這里的Decode和Netty本身的decode不一樣镊绪，Netty本身的decode在執(zhí)行NettyHandler之前就執(zhí)行解碼了匀伏；后續(xù)的操作在Exchange層進(jìn)行處理，本文暫時先不做介紹蝴韭；
2.Client端
同樣查看父類AbstractClient够颠，構(gòu)造方法如下：
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()));
}
客戶端需要提供重連機(jī)制，所以初始化的幾個參數(shù)都和重連有關(guān)榄鉴，send_reconnect表示在發(fā)送消息時發(fā)現(xiàn)連接已經(jīng)斷開是否發(fā)起重連履磨，reconnect_warning_period表示多久報一次重連警告蛉抓，shutdown_timeout表示連接服務(wù)器一直連接不上的超時時間；接下來就是調(diào)用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客戶端的常規(guī)代碼巷送，設(shè)置了和Server端相同的NettyHandler，decoder和encoder综苔；下面重點(diǎn)看看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();
}
}
首先判定是否已經(jīng)連接惩系，如果連接直接return；接下來初始化連接狀態(tài)檢查器如筛，定期檢查channel是否連接堡牡，連接斷開會進(jìn)行重連操作，具體代碼如下：在此我向大家推薦一個架構(gòu)學(xué)習(xí)交流群杨刨。交流學(xué)習(xí)群號：821169538 ?里面會分享一些資深架構(gòu)師錄制的視頻錄像：有Spring晤柄，MyBatis，Netty源碼分析妖胀，高并發(fā)芥颈、高性能、分布式赚抡、微服務(wù)架構(gòu)的原理爬坑，JVM性能優(yōu)化、分布式架構(gòu)等這些成為架構(gòu)師必備的知識體系涂臣。還能領(lǐng)取免費(fèi)的學(xué)習(xí)資源盾计，目前受益良多。
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);
}
}
創(chuàng)建了一個Runnable赁遗，用來檢測是否連接署辉，如果連接斷開，調(diào)用connect方法岩四；定時調(diào)度交給ScheduledThreadPoolExecutor來執(zhí)行哭尝；初始化之后就調(diào)用具體Client的doConnect操作，也是通訊框架的一些常規(guī)代碼剖煌，此處不列出了材鹦；后續(xù)關(guān)于NettyChannel的介紹和Server端類似，不過多進(jìn)行介紹耕姊；
總結(jié)
本文重點(diǎn)分析了dubbo架構(gòu)中的transport層侠姑，具體圍繞Transporter, Client, Server，ChannelHandler幾個類展開箩做，關(guān)于后續(xù)的處理將在exchange信息交換層；
