下面我们来分析一下Dubbo的Transport层。我们之前说过，transport层抽象mina和netty为统一接口，以Message为中心，扩展接口为Channel、Transporter、Client、Server和Codec。也就是通过netty进行网络通信。先看一下Transporter接口
/**
* Bind a server.
*
* 绑定一个服务器
*
* @param url server url
* @param handler 通道处理器
* @return server 服务器
* @throws RemotingException 当绑定发生异常时
* @see com.alibaba.dubbo.remoting.Transporters#bind(URL, Receiver, ChannelHandler)
*/
@Adaptive({Constants.SERVER_KEY, Constants.TRANSPORTER_KEY})
Server bind(URL url, ChannelHandler handler) throws RemotingException;
/**
* Connect to a server.
*
* 连接一个服务器，即创建一个客户端
*
* @param url server url 服务器地址
* @param handler 通道处理器
* @return client 客户端
* @throws RemotingException 当连接发生异常时
* @see com.alibaba.dubbo.remoting.Transporters#connect(URL, Receiver, ChannelListener)
*/
@Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})
Client connect(URL url, ChannelHandler handler) throws RemotingException;
再看一下NettyTransporter这个类
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);
}
}
用NettyServer和NettyClient封装了netty的网络通信功能。
先看下服务端。先来看一下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方法通过具体的通讯框架绑定端口启动服务
看一下NettyServer的实现
protected void doOpen() {
// 设置日志工厂
NettyHelper.setNettyLoggerFactory();
// 创建线程池
ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
// 创建 ChannelFactory 对象
ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
// 实例化 ServerBootstrap
bootstrap = new ServerBootstrap(channelFactory);
// 创建 NettyHandler 对象
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
// 设置 `channels` 属性
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.setPipelineFactory(new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() {
// 创建 NettyCodecAdapter 对象
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());
}
再来看一下客户端。先来看一下AbstractClient这个类
public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
super(url, handler);
// 从 URL 中，获得重连相关配置项
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()方法。看一下NettyClient的实现
protected void doOpen() {
// 设置日志工厂
NettyHelper.setNettyLoggerFactory();
// 实例化 ServerBootstrap
bootstrap = new ClientBootstrap(channelFactory);
// 设置可选项
// config
// @see org.jboss.netty.channel.socket.SocketChannelConfig
bootstrap.setOption("keepAlive", true);
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("connectTimeoutMillis", getTimeout());
// 创建 NettyHandler 对象
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
// 设置责任链路
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() {
// 创建 NettyCodecAdapter 对象
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;
}
});
}
Dubbo的Transport层就分析到这里了。
