image.png
如上所示，在dubbo的自介里面，是这样介绍exchange层与transport层的
exchange 信息交换层：封装请求响应模式，同步转异步，以 Request, Response 为中心，扩展接口为 Exchanger, ExchangeChannel, ExchangeClient, ExchangeServer
transport 网络传输层：抽象 mina 和 netty 为统一接口，以 Message 为中心，扩展接口为 Channel, Transporter, Client, Server, Codec
exchange封装语义信息，等于说操作的是Request Response的语义封装，而transport层做为传输层是不care任何的语义信息的，它只负责单向的数据传输。
为了理解上面所述，我们先看下dubbo里面提供的Endpoint接口。
public interface Endpoint {
//当前Endpoint对应的url（这个url是dubbo提供的一个信息载体）
URL getUrl();
//当前Endpoint对应的ChannelHandler
ChannelHandler getChannelHandler();
//当前Endpoint对应的本地地址
InetSocketAddress getLocalAddress();
//使用Endpoint发送消息
void send(Object message) throws RemotingException;
//与上相同，添加的sent用于幂等校验
void send(Object message, boolean sent) throws RemotingException;
//关闭Endpoint
void close();
//在timeout时间内关闭Endpoint
void close(int timeout);
//开始关闭Endpoint
void startClose();
//Endpoint是否已经关闭
boolean isClosed();
}
dubbo为什么抽象出Endpoint的概念，因为对于Client和Service来说，其实都是一个Endpoint，只是语义上的不同，两者在物理上是p2p的概念，所以用Endpoint来统一抽象一下。如下
image.png
Client和Service都extends于Endpoint，这里有个很关键的概念要说下，dubbo里面的Client到底是个啥，Client就是对应一个虚拟机吗？如果你这样理解的话，会有很多不理解的地方，在dubbo里面，每个DubboInvoker都内置了一个ExchangeClient[] clients;默认情况下是一个client，所以如果refer的dubbo Provider 服务处于不同的虚拟机的话，一个虚拟机里面会有多个Client（根据refer的地址来缓存共享Client）。
所以看到上面的图我们就明白了，为啥Client是继承Channel的，因为一个Client只关联到一个远程的Provider的server，那么其关联的Channel也就是一对一的，所以我们可以这样的继承和实现。
我们接着看下Channel接口的实现，Channel是连接p2p的通道，
public interface Channel extends Endpoint {
//拿到远程地址，从这个方法的语义出发，我们可以看大Channel是关联到Client上的
InetSocketAddress getRemoteAddress();
//是否已经连接，那么肯定是针对服务端来说的，就是是否已经连接上了服务端
boolean isConnected();
boolean hasAttribute(String key);
Object getAttribute(String key);
void setAttribute(String key, Object value);
void removeAttribute(String key);
}
关于Attribute的几个简单方法我们先不谈，从Channel的方法的语义上分析，我们可以定位到Channel是为Client而准备的。
故而我们可以看到Client直接的实现了对Channel的继承，源码如下
public interface Client extends Endpoint, Channel, Resetable {
//从新连接
void reconnect() throws RemotingException;
@Deprecated
void reset(com.alibaba.dubbo.common.Parameters parameters);
}
我们可以看到Client只是新增了一个reconnect方法，因为默认是使用tcp来打开keepalive连接的，那么如果发生网络抖动，我们是需要重连的，保证通道的可用。
而接口Resetable，表明实现类可以通过url进行属性的重新设置，在前面说过，dubbo里面的peer端都是通过url这个对象进行属性的更新和设置的。
public interface Resetable {
void reset(URL url);
}
我们再看下Server接口的定义
public interface Server extends Endpoint, Resetable {
//是否已经绑定到本地，也就是是否启动成功，因为是Server，所以不用connect到远程
//只需要监听bind的本地port就可以
boolean isBound();
//拿到所有连接到此Server的Channel，可以认为这个Channel是对Client的抽象
Collection<Channel> getChannels();
//根据远程地址（也就是Client的地址）返回绑定的Channel
Channel getChannel(InetSocketAddress remoteAddress);
@Deprecated //作废，不介绍，使用Resetable里面的reset方法。
void reset(com.alibaba.dubbo.common.Parameters parameters);
}
由于exchange 信息交换层：封装请求响应模式，同步转异步，以 Request, Response 为中心，扩展接口为 Exchanger, ExchangeChannel, ExchangeClient, ExchangeServer
我们通过看前面的接口发现Channel只有单向send能力（继承于Endpoint接口），而ExchangeChannel位于exchange层，具有了接收response的能力，我们看接口源码
public interface ExchangeChannel extends Channel {
//抽象出request方法，在Endpont的简单的send的基础上抽象出了request和response的能力
ResponseFuture request(Object request) throws RemotingException;
//带超时时间的request方法
ResponseFuture request(Object request, int timeout) throws RemotingException;
//后面再分析这个
ExchangeHandler getExchangeHandler();
//指定时间内关闭这个close。
void close(int timeout);
}
而针对 Channel的不同的状态事件，我们绑定ChannelHandler来进行逻辑上的处理，接口定义如下
@SPI
public interface ChannelHandler {
//当channel connected时的处理方法
void connected(Channel channel) throws RemotingException;
//当channel disconnected时的处理方法
void disconnected(Channel channel) throws RemotingException;
//当channel send message时的处理方法
void sent(Channel channel, Object message) throws RemotingException;
//当 channel received message时的处理方法
void received(Channel channel, Object message) throws RemotingException;
//当channel caught exception时的处理方法
void caught(Channel channel, Throwable exception) throws RemotingException;
}
ChannelHandler不关心语义的处理，而ExchangeHandler关心语义，所以其添加了reply方法，如下
public interface ExchangeHandler extends ChannelHandler, TelnetHandler {
//其中的request是具有语义的request或是response
Object reply(ExchangeChannel channel, Object request) throws RemotingException;
}
好，看了如上之后，我们看下transport层的Channel的继承体系，如下
image.png
可以看到transport的Channel对Netty，Grizzly，Mina的Channel做了统一的封装，这样用户可以灵活的使用spi进行配置。
我们看下AbstractChannel的源码如下
public abstract class AbstractChannel extends AbstractPeer implements Channel {
public AbstractChannel(URL url, ChannelHandler handler) {
super(url, handler);
}
@Override
public void send(Object message, boolean sent) throws RemotingException {
if (isClosed()) {
throw new RemotingException(this, "Failed to send message "
+ (message == null ? "" : message.getClass().getName()) + ":" + message
+ ", cause: Channel closed. channel: " + getLocalAddress() + " -> " + getRemoteAddress());
}
}
@Override
public String toString() {
return getLocalAddress() + " -> " + getRemoteAddress();
}
}
基本没有实现什么功能，只是简单的做了一个Channel的状态的判断。所有的实现都由子类来实现。AbstractPeer只是为了抽象出p2p的概念，很简单，不用介绍。
由于最出名的nio框架还是netty，我们选取NettyChannel进行分析，如下
final class NettyChannel extends AbstractChannel {
private static final Logger logger = LoggerFactory.getLogger(NettyChannel.class);
//这里封装了netty的Channel与dubbo自己封装的NettyChannel的缓存关系
private static final ConcurrentMap<Channel, NettyChannel> channelMap = new ConcurrentHashMap<Channel, NettyChannel>();
//这个是实际引用的netty的Channel
private final Channel channel;
//属性
private final Map<String, Object> attributes = new ConcurrentHashMap<String, Object>();
//构造函数为私有，所以只能自己调用
private NettyChannel(Channel channel, URL url, ChannelHandler handler) {
super(url, handler);
if (channel == null) {
throw new IllegalArgumentException("netty channel == null;");
}
this.channel = channel; //典型的适配器模式，将netty的Channel适配成了dubbo的NettyChannel
}
//将netty的Channel封装成dubbo的NettyChannel
static NettyChannel getOrAddChannel(Channel ch, URL url, ChannelHandler handler) {
if (ch == null) {
return null;
}
NettyChannel ret = channelMap.get(ch);
if (ret == null) {
NettyChannel nettyChannel = new NettyChannel(ch, url, handler);
if (ch.isActive()) {
ret = channelMap.putIfAbsent(ch, nettyChannel);
}
if (ret == null) {
ret = nettyChannel;
}
}
return ret;
}
//如果Channel Disconnected ，对缓存进行移除
static void removeChannelIfDisconnected(Channel ch) {
if (ch != null && !ch.isActive()) {
channelMap.remove(ch);
}
}
//典型的代理模式
public InetSocketAddress getLocalAddress() {
return (InetSocketAddress) channel.localAddress();
}
//拿到远程Server的地址
public InetSocketAddress getRemoteAddress() {
return (InetSocketAddress) channel.remoteAddress();
}
//是否Connected
public boolean isConnected() {
return !isClosed() && channel.isActive();
}
@Override
public void send(Object message, boolean sent) throws RemotingException {
super.send(message, sent);
boolean success = true;
int timeout = 0;
try {
ChannelFuture future = channel.writeAndFlush(message);
if (sent) {
timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
success = future.await(timeout);
}
Throwable cause = future.cause();
if (cause != null) {
throw cause;
}
} catch (Throwable e) {
throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress() + ", cause: " + e.getMessage(), e);
}
if (!success) {
throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress()
+ "in timeout(" + timeout + "ms) limit");
}
}
@Override
public void close() {
try {
super.close();
} catch (Exception e) {
logger.warn(e.getMessage(), e);
}
try {
removeChannelIfDisconnected(channel);
} catch (Exception e) {
logger.warn(e.getMessage(), e);
}
try {
attributes.clear();
} catch (Exception e) {
logger.warn(e.getMessage(), e);
}
try {
if (logger.isInfoEnabled()) {
logger.info("Close netty channel " + channel);
}
channel.close();
} catch (Exception e) {
logger.warn(e.getMessage(), e);
}
}
@Override
public boolean hasAttribute(String key) {
return attributes.containsKey(key);
}
@Override
public Object getAttribute(String key) {
return attributes.get(key);
}
@Override
public void setAttribute(String key, Object value) {
if (value == null) { // The null value unallowed in the ConcurrentHashMap.
attributes.remove(key);
} else {
attributes.put(key, value);
}
}
@Override
public void removeAttribute(String key) {
attributes.remove(key);
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + ((channel == null) ? 0 : channel.hashCode());
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj) return true;
if (obj == null) return false;
if (getClass() != obj.getClass()) return false;
NettyChannel other = (NettyChannel) obj;
if (channel == null) {
if (other.channel != null) return false;
} else if (!channel.equals(other.channel)) return false;
return true;
}
@Override
public String toString() {
return "NettyChannel [channel=" + channel + "]";
}
}
如上可以看到NettyChannel的能力都是借用了netty的Channel的能力。NettyChannel只是做了简单的封装。
由于在客户端，Channel与Client一一对应，我们看下transport层的Client的继承体系，如下
image.png
从继承关系看由于AbstractClient继承于ChannelHandler，所以其拥有了自我handler的能力。
我们先看下AbstractPeer的源码如下
public abstract class AbstractPeer implements Endpoint, ChannelHandler {
//代理的ChannelHandler
private final ChannelHandler handler;
//url就是配置信息
private volatile URL url;
//是否在关闭中
private volatile boolean closing;
//是否已经关闭
private volatile boolean closed;
public AbstractPeer(URL url, ChannelHandler handler) {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handler == null) {
throw new IllegalArgumentException("handler == null");
}
this.url = url;
this.handler = handler;
}
@Override
public void send(Object message) throws RemotingException {
send(message, url.getParameter(Constants.SENT_KEY, false));
}
@Override
public void close() {
closed = true;
}
@Override
public void close(int timeout) {
close();
}
@Override
public void startClose() {
if (isClosed()) {
return;
}
closing = true;
}
@Override
public URL getUrl() {
return url;
}
protected void setUrl(URL url) {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
this.url = url;
}
@Override
public ChannelHandler getChannelHandler() {
if (handler instanceof ChannelHandlerDelegate) {
return ((ChannelHandlerDelegate) handler).getHandler();
} else {
return handler;
}
}
/**
* @return ChannelHandler
*/
@Deprecated
public ChannelHandler getHandler() {
return getDelegateHandler();
}
/**
* Return the final handler (which may have been wrapped). This method should be distinguished with getChannelHandler() method
*
* @return ChannelHandler
*/
public ChannelHandler getDelegateHandler() {
return handler;
}
@Override
public boolean isClosed() {
return closed;
}
public boolean isClosing() {
return closing && !closed;
}
@Override
public void connected(Channel ch) throws RemotingException {
if (closed) {
return;
}
handler.connected(ch);
}
@Override
public void disconnected(Channel ch) throws RemotingException {
handler.disconnected(ch);
}
@Override
public void sent(Channel ch, Object msg) throws RemotingException {
if (closed) {
return;
}
handler.sent(ch, msg);
}
@Override
public void received(Channel ch, Object msg) throws RemotingException {
if (closed) {
return;
}
handler.received(ch, msg);
}
@Override
public void caught(Channel ch, Throwable ex) throws RemotingException {
handler.caught(ch, ex);
}
}
如上可以看到AbstractPeer内部引用了一个ChannelHandler来实现ChannelHandler的功能，典型的装饰模式。
我们再看看AbstractEndpoint的源码如下
public abstract class AbstractEndpoint extends AbstractPeer implements Resetable {
private static final Logger logger = LoggerFactory.getLogger(AbstractEndpoint.class);
//内部封装的编解码器，对于transport层的Endpoint来说，我操作的都是字符数组，怎么编解码需要Codec2来负责黏包，半包的问题。
private Codec2 codec;
//超时时间
private int timeout;
//链接超时时间
private int connectTimeout;
public AbstractEndpoint(URL url, ChannelHandler handler) {
super(url, handler);
//通过url拿到自定义的（如果有的话）的codec
this.codec = getChannelCodec(url);
//通过url拿到自定义的（如果有的话）的timeout
this.timeout = url.getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
// //通过url拿到自定义的（如果有的话）的connectTimeout
this.connectTimeout = url.getPositiveParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT);
}
protected static Codec2 getChannelCodec(URL url) {
String codecName = url.getParameter(Constants.CODEC_KEY, "telnet");
if (ExtensionLoader.getExtensionLoader(Codec2.class).hasExtension(codecName)) {
return ExtensionLoader.getExtensionLoader(Codec2.class).getExtension(codecName);
} else {
return new CodecAdapter(ExtensionLoader.getExtensionLoader(Codec.class)
.getExtension(codecName));
}
}
//根据url重置timeout与connectTimeout属性
@Override
public void reset(URL url) {
if (isClosed()) {
throw new IllegalStateException("Failed to reset parameters "
+ url + ", cause: Channel closed. channel: " + getLocalAddress());
}
try {
if (url.hasParameter(Constants.TIMEOUT_KEY)) {
int t = url.getParameter(Constants.TIMEOUT_KEY, 0);
if (t > 0) {
this.timeout = t;
}
}
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
try {
if (url.hasParameter(Constants.CONNECT_TIMEOUT_KEY)) {
int t = url.getParameter(Constants.CONNECT_TIMEOUT_KEY, 0);
if (t > 0) {
this.connectTimeout = t;
}
}
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
try {
if (url.hasParameter(Constants.CODEC_KEY)) {
this.codec = getChannelCodec(url);
}
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
}
@Deprecated
public void reset(com.alibaba.dubbo.common.Parameters parameters) {
reset(getUrl().addParameters(parameters.getParameters()));
}
protected Codec2 getCodec() {
return codec;
}
protected int getTimeout() {
return timeout;
}
protected int getConnectTimeout() {
return connectTimeout;
}
}
如上可以看到AbstractEndpoint封装了Codec2和两个超时的属性。
我们再看下AbstractClient。
public abstract class AbstractClient extends AbstractEndpoint implements Client {
//对应的线程池的名字的前缀
protected static final String CLIENT_THREAD_POOL_NAME = "DubboClientHandler";
private static final Logger logger = LoggerFactory.getLogger(AbstractClient.class);
//对应的线程池的名字id
private static final AtomicInteger CLIENT_THREAD_POOL_ID = new AtomicInteger();
//定时任务，用于client的connect状态的检测和重连
private static final ScheduledThreadPoolExecutor reconnectExecutorService = new ScheduledThreadPoolExecutor(2, new NamedThreadFactory("DubboClientReconnectTimer", true));
//connect时的锁
private final Lock connectLock = new ReentrantLock();
//在send的时候，如果send_reconnect为ture，那么会检查下client的状态，如果是disconnect状态，那么会发起连接
private final boolean send_reconnect;
//重连的次数的统计
private final AtomicInteger reconnect_count = new AtomicInteger(0);
// Reconnection error log has been called before?
private final AtomicBoolean reconnect_error_log_flag = new AtomicBoolean(false);
// reconnect warning period. Reconnect warning interval (log warning after how many times) //for test
private final int reconnect_warning_period;
private final long shutdown_timeout;
//client对应的线程池
protected volatile ExecutorService executor;
//重连返回的结果的占位符Future
private volatile ScheduledFuture<?> reconnectExecutorFuture = null;
// the last successed connected time
private long lastConnectedTime = System.currentTimeMillis();
public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
super(url, handler);
//默认是false
send_reconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, false);
//默认15分钟
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);
}
//得到这个Client关联的线程池
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()));
}
protected static ChannelHandler wrapChannelHandler(URL url, ChannelHandler handler) {
url = ExecutorUtil.setThreadName(url, CLIENT_THREAD_POOL_NAME);
url = url.addParameterIfAbsent(Constants.THREADPOOL_KEY, Constants.DEFAULT_CLIENT_THREADPOOL);
return ChannelHandlers.wrap(handler, url);
}
/**
* @param url
* @return 0-false Reconnect间隔时间
*/
private static int getReconnectParam(URL url) {
int reconnect;
String param = url.getParameter(Constants.RECONNECT_KEY);
if (param == null || param.length() == 0 || "true".equalsIgnoreCase(param)) {
reconnect = Constants.DEFAULT_RECONNECT_PERIOD;
} else if ("false".equalsIgnoreCase(param)) {
reconnect = 0;
} else {
try {
reconnect = Integer.parseInt(param);
} catch (Exception e) {
throw new IllegalArgumentException("reconnect param must be nonnegative integer or false/true. input is:" + param);
}
if (reconnect < 0) {
throw new IllegalArgumentException("reconnect param must be nonnegative integer or false/true. input is:" + param);
}
}
return reconnect;
}
/**
* init reconnect thread 这个重连线程会在client断开的时候发起重连
*/
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);
}
}
//不进行重连的定时操作
private synchronized void destroyConnectStatusCheckCommand() {
try {
if (reconnectExecutorFuture != null && !reconnectExecutorFuture.isDone()) {
reconnectExecutorFuture.cancel(true);
reconnectExecutorService.purge();
}
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
}
//创建新的线程池
protected ExecutorService createExecutor() {
return Executors.newCachedThreadPool(new NamedThreadFactory(CLIENT_THREAD_POOL_NAME + CLIENT_THREAD_POOL_ID.incrementAndGet() + "-" + getUrl().getAddress(), true));
}
public InetSocketAddress getConnectAddress() {
return new InetSocketAddress(NetUtils.filterLocalHost(getUrl().getHost()), getUrl().getPort());
}
@Override
public InetSocketAddress getRemoteAddress() {
Channel channel = getChannel();
if (channel == null)
return getUrl().toInetSocketAddress();
return channel.getRemoteAddress();
}
@Override
public InetSocketAddress getLocalAddress() {
Channel channel = getChannel();
if (channel == null)
return InetSocketAddress.createUnresolved(NetUtils.getLocalHost(), 0);
return channel.getLocalAddress();
}
//还是抛给了channel去判断
@Override
public boolean isConnected() {
Channel channel = getChannel();
if (channel == null)
return false;
return channel.isConnected();
}
@Override
public Object getAttribute(String key) {
Channel channel = getChannel();
if (channel == null)
return null;
return channel.getAttribute(key);
}
@Override
public void setAttribute(String key, Object value) {
Channel channel = getChannel();
if (channel == null)
return;
channel.setAttribute(key, value);
}
@Override
public void removeAttribute(String key) {
Channel channel = getChannel();
if (channel == null)
return;
channel.removeAttribute(key);
}
@Override
public boolean hasAttribute(String key) {
Channel channel = getChannel();
if (channel == null)
return false;
return channel.hasAttribute(key);
}
@Override
public void send(Object message, boolean sent) throws RemotingException {
//如果send_reconnect为true，那么进行状态检测，如果状态是非连接态，进行重连
if (send_reconnect && !isConnected()) {
connect();
}
//留给子类实现
Channel channel = getChannel();
//TODO Can the value returned by getChannel() be null? need improvement.
if (channel == null || !channel.isConnected()) {
throw new RemotingException(this, "message can not send, because channel is closed . url:" + getUrl());
}
//通过channel将消息发出去
channel.send(message, sent);
}
//发起连接
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: " + getConnectTimeout() + "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();
}
}
public void disconnect() {
connectLock.lock();
try {
destroyConnectStatusCheckCommand();
try {
Channel channel = getChannel();
if (channel != null) {
channel.close();
}
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
try {
//核心方法由子类实现
doDisConnect();
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
} finally {
connectLock.unlock();
}
}
//提供重连的方法，先断开，再连接
@Override
public void reconnect() throws RemotingException {
disconnect();
connect();
}
//close的时候先销毁外围资源
@Override
public void close() {
try {
if (executor != null) {
ExecutorUtil.shutdownNow(executor, 100);
}
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
try {
super.close();
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
try {
disconnect();
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
try {
doClose();
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
}
@Override
public void close(int timeout) {
ExecutorUtil.gracefulShutdown(executor, timeout);
close();
}
@Override
public String toString() {
return getClass().getName() + " [" + getLocalAddress() + " -> " + getRemoteAddress() + "]";
}
/**
* Open client.
*
* @throws Throwable
*/
protected abstract void doOpen() throws Throwable;
/**
* Close client.
*
* @throws Throwable
*/
protected abstract void doClose() throws Throwable;
/**
* Connect to server.
*
* @throws Throwable
*/
protected abstract void doConnect() throws Throwable;
/**
* disConnect to server.
*
* @throws Throwable
*/
protected abstract void doDisConnect() throws Throwable;
/**
* Get the connected channel.
*
* @return channel
*/
protected abstract Channel getChannel();
}
如上我们可以看到AbstractClient对Client的链接态做了判断和重连机制，所有的核心实现还是由子类来实现我们看下子类是怎么实现的，由于netty最为有名，我们选取NettyClient做分析。
源码如下
public class NettyClient extends AbstractClient {
private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);
private static final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(Constants.DEFAULT_IO_THREADS, new DefaultThreadFactory("NettyClientWorker", true));
//netty客户端启动类
private Bootstrap bootstrap;
//volatile,使用引用的方式进行使用
private volatile Channel channel; // volatile, please copy reference to use
public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException {
//wrapChannelHandler我们后面分析
super(url, wrapChannelHandler(url, handler));
}
//开启客户端，其实就是对Client进行设置，为connect前做准备
@Override
protected void doOpen() throws Throwable {
final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this);
bootstrap = new Bootstrap();
bootstrap.group(nioEventLoopGroup)
.option(ChannelOption.SO_KEEPALIVE, true)
.option(ChannelOption.TCP_NODELAY, true)
//缓存使用池化技术
.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
//.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())
.channel(NioSocketChannel.class);
//链接超时时间设置成3秒之上
if (getConnectTimeout() < 3000) {
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
} else {
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout());
}
bootstrap.handler(new ChannelInitializer() {
@Override
protected void initChannel(Channel ch) throws Exception {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
//设置编码器
.addLast("decoder", adapter.getDecoder())
//设置解码器
.addLast("encoder", adapter.getEncoder())
//设置handler，其实就是本身
.addLast("handler", nettyClientHandler);
}
});
}
@Override
protected void doConnect() throws Throwable {
long start = System.currentTimeMillis();
ChannelFuture future = bootstrap.connect(getConnectAddress());
try {
boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
//发起连接
if (ret && future.isSuccess()) {
Channel newChannel = future.channel();
try {
//销毁原来关联的老的channel
// Close old channel
Channel oldChannel = NettyClient.this.channel; // copy reference
if (oldChannel != null) {
try {
if (logger.isInfoEnabled()) {
logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
}
oldChannel.close();
} finally {
NettyChannel.removeChannelIfDisconnected(oldChannel);
}
}
} finally {
if (NettyClient.this.isClosed()) {
try {
if (logger.isInfoEnabled()) {
logger.info("Close new netty channel " + newChannel + ", because the client closed.");
}
newChannel.close();
} finally {
NettyClient.this.channel = null;
NettyChannel.removeChannelIfDisconnected(newChannel);
}
} else {
//将新生成的channl赋值给当前，这样就完成了整个的初始化
NettyClient.this.channel = newChannel;
}
}
} else if (future.cause() != null) {
throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
+ getRemoteAddress() + ", error message is:" + future.cause().getMessage(), future.cause());
} else {
throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
+ getRemoteAddress() + " client-side timeout "
+ getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client "
+ NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion());
}
} finally {
if (!isConnected()) {
//future.cancel(true);
}
}
}
//关闭连接，其实就是移除缓存，为啥不把这个nioEventLoopGroup也干掉呢，因为这个是共享的，别人还在用
@Override
protected void doDisConnect() throws Throwable {
try {
NettyChannel.removeChannelIfDisconnected(channel);
} catch (Throwable t) {
logger.warn(t.getMessage());
}
}
@Override
protected void doClose() throws Throwable {
//can't shutdown nioEventLoopGroup
//nioEventLoopGroup.shutdownGracefully();
}
//返回当前的channel并将其缓存并封装成dubbo的channel进行返回
@Override
protected com.alibaba.dubbo.remoting.Channel getChannel() {
Channel c = channel;
if (c == null || !c.isActive())
return null;
return NettyChannel.getOrAddChannel(c, getUrl(), this);
}
}
如上的代码还是比较容易的。如上是对transport层的client的分析，对于transport来说没有任何的业务含义在里面，所以接下来我们继续的分析exchange层的client。
整个exchange层的client的继承关系如下
image.png
前面我们分析过，Client与Channel是一一对应的，所以dubbo抽象出了个接口ExchangeClient，源码如下
public interface ExchangeClient extends Client, ExchangeChannel {
}
而例外一个实现类HeaderExchangeChannel是在exchange层具有业务语义的Channel,我们看下其源码实现。
final class HeaderExchangeChannel implements ExchangeChannel {
private static final Logger logger = LoggerFactory.getLogger(HeaderExchangeChannel.class);
private static final String CHANNEL_KEY = HeaderExchangeChannel.class.getName() + ".CHANNEL";
//内部封装的channel，我们可以看到这是个装饰模式，给没有业务语义的channel
//添加业务语义
private final Channel channel;
private volatile boolean closed = false;
HeaderExchangeChannel(Channel channel) {
if (channel == null) {
throw new IllegalArgumentException("channel == null");
}
this.channel = channel;
}
//
static HeaderExchangeChannel getOrAddChannel(Channel ch) {
if (ch == null) {
return null;
}
HeaderExchangeChannel ret = (HeaderExchangeChannel) ch.getAttribute(CHANNEL_KEY);
if (ret == null) {
ret = new HeaderExchangeChannel(ch);
if (ch.isConnected()) {
ch.setAttribute(CHANNEL_KEY, ret);
}
}
return ret;
}
static void removeChannelIfDisconnected(Channel ch) {
if (ch != null && !ch.isConnected()) {
ch.removeAttribute(CHANNEL_KEY);
}
}
@Override
public void send(Object message) throws RemotingException {
send(message, getUrl().getParameter(Constants.SENT_KEY, false));
}
@Override
public void send(Object message, boolean sent) throws RemotingException {
if (closed) {
throw new RemotingException(this.getLocalAddress(), null, "Failed to send message " + message + ", cause: The channel " + this + " is closed!");
}
//如果message是Request 或者 Response 或者String（兼容telnet）直接的发送
if (message instanceof Request
|| message instanceof Response
|| message instanceof String) {
channel.send(message, sent);
} else {
//不然将其封装成Request进行发送
Request request = new Request();
request.setVersion(Version.getProtocolVersion());
//单向发送，不关注返回结果，即时的返回即可
request.setTwoWay(false);
request.setData(message);
channel.send(request, sent);
}
}
//发送，并拿到返回结果，这里就有了response的信息了
@Override
public ResponseFuture request(Object request) throws RemotingException {
return request(request, channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
}
@Override
public ResponseFuture request(Object request, int timeout) throws RemotingException {
if (closed) {
throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
}
// create request.
Request req = new Request();
req.setVersion(Version.getProtocolVersion());
//需要返回结果，会进行超时等待返回结果
req.setTwoWay(true);
req.setData(request);
DefaultFuture future = new DefaultFuture(channel, req, timeout);
try {
channel.send(req);
} catch (RemotingException e) {
future.cancel();
throw e;
}
return future;
}
@Override
public boolean isClosed() {
return closed;
}
@Override
public void close() {
try {
channel.close();
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
}
// graceful close
@Override
public void close(int timeout) {
if (closed) {
return;
}
closed = true;
if (timeout > 0) {
long start = System.currentTimeMillis();
while (DefaultFuture.hasFuture(channel)
&& System.currentTimeMillis() - start < timeout) {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
logger.warn(e.getMessage(), e);
}
}
}
close();
}
@Override
public void startClose() {
channel.startClose();
}
@Override
public InetSocketAddress getLocalAddress() {
return channel.getLocalAddress();
}
@Override
public InetSocketAddress getRemoteAddress() {
return channel.getRemoteAddress();
}
@Override
public URL getUrl() {
return channel.getUrl();
}
@Override
public boolean isConnected() {
return channel.isConnected();
}
@Override
public ChannelHandler getChannelHandler() {
return channel.getChannelHandler();
}
@Override
public ExchangeHandler getExchangeHandler() {
return (ExchangeHandler) channel.getChannelHandler();
}
@Override
public Object getAttribute(String key) {
return channel.getAttribute(key);
}
@Override
public void setAttribute(String key, Object value) {
channel.setAttribute(key, value);
}
@Override
public void removeAttribute(String key) {
channel.removeAttribute(key);
}
@Override
public boolean hasAttribute(String key) {
return channel.hasAttribute(key);
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + ((channel == null) ? 0 : channel.hashCode());
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj) return true;
if (obj == null) return false;
if (getClass() != obj.getClass()) return false;
HeaderExchangeChannel other = (HeaderExchangeChannel) obj;
if (channel == null) {
if (other.channel != null) return false;
} else if (!channel.equals(other.channel)) return false;
return true;
}
@Override
public String toString() {
return channel.toString();
}
}
如上我们可以看到HeaderExchangeChannel已经有了Request和Response的业务语义了。
即时是在exchange层，Channel也是底层的实现，我们业务需要关心的是Client。
我们看下ExchangeClient的几个实现。
ReferenceCountExchangeClient略过，只是简单的记录对某个ExchangeClient的引用。
HeaderExchangeClient底层也是使用了ExchangeChannel但是拥有了heartbeat的能力。
LazyConnectExchangeClient 只有在真正的发起连接的时候，才初始化客户端。
这3个Client其实都是针对Channel的封装，使用很简单，我们就不说了。
最后，我们看下exchange与transport层是如何关联起来的，其实在LazyConnectExchangeClient已经有了，其有个initClient方法，我们看下
private void initClient() throws RemotingException {
if (client != null)
return;
if (logger.isInfoEnabled()) {
logger.info("Lazy connect to " + url);
}
connectLock.lock();
try {
if (client != null)
return;
//这一句是重点
this.client = Exchangers.connect(url, requestHandler);
} finally {
connectLock.unlock();
}
}
我们再看Exchangers里面的connect方法，如下
public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handler == null) {
throw new IllegalArgumentException("handler == null");
}
url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
return getExchanger(url).connect(url, handler);
//
public static Exchanger getExchanger(String type) {
return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);
}
Exchanger其实就是对exchange层的最高级的封装，如下
public interface Exchanger {
//生成服务端的ExchangeServer
@Adaptive({Constants.EXCHANGER_KEY})
ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException;
//生成客户端的ExchangeClient
@Adaptive({Constants.EXCHANGER_KEY})
ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException;
}
其具体实现是HeaderExchanger
如下
public class HeaderExchanger implements Exchanger {
public static final String NAME = "header";
@Override
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
//返回的就是我们前面讲的HeaderExchangeClient
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))));
}
}
照这样分析，ExchangeClient应该是封装了一个Transporter层的client，果不其然，
我们看
Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler)))
的具体返回。
Transporters是对Transporter层的最高级封装，如下
public class Transporters {
static {
// check duplicate jar package 防痴呆设计，防止重复的jar冲突
Version.checkDuplicate(Transporters.class);
Version.checkDuplicate(RemotingException.class);
}
private Transporters() {
}
public static Client connect(String url, ChannelHandler... handler) throws RemotingException {
return connect(URL.valueOf(url), handler);
}
public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
ChannelHandler handler;
if (handlers == null || handlers.length == 0) {
handler = new ChannelHandlerAdapter();
} else if (handlers.length == 1) {
handler = handlers[0];
} else {
//
handler = new ChannelHandlerDispatcher(handlers);
}
return getTransporter().connect(url, handler);
}
public static Transporter getTransporter() {
return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
}
}
而getTransporter方法，默认返回的就是我们前面分析的NettyClient，这样整个就串起来。
