Dubbo 源码分析 - 服务调用过程 (5)

到这里大家终于看到了 Request 语义了，上面的方法首先定义了一个 Request 对象，然后在将该对象传给 NettyClient 的 send 方法，进行后续的调用。需要说明的是，NettyClient 中并未实现 send 方法，该方法继承自父类 AbstractPeer，下面直接分析 AbstractPeer 的代码。

public abstract class AbstractPeer implements Endpoint, ChannelHandler { @Override public void send(Object message) throws RemotingException { // 该方法由 AbstractClient 类实现 send(message, url.getParameter(Constants.SENT_KEY, false)); } // 省略其他方法 } public abstract class AbstractClient extends AbstractEndpoint implements Client { @Override public void send(Object message, boolean sent) throws RemotingException { if (send_reconnect && !isConnected()) { connect(); } // 获取 Channel，getChannel 是一个抽象方法，具体由子类实现 Channel channel = getChannel(); if (channel == null || !channel.isConnected()) { throw new RemotingException(this, "message can not send ..."); } // 继续向下调用 channel.send(message, sent); } protected abstract Channel getChannel(); // 省略其他方法 }

默认情况下，Dubbo 使用 Netty 作为底层的通信框架，因此下面我们到 NettyClient 类中看一下 getChannel 方法的实现逻辑。

public class NettyClient extends AbstractClient { // 这里的 Channel 全限定名称为 org.jboss.netty.channel.Channel private volatile Channel channel; @Override protected com.alibaba.dubbo.remoting.Channel getChannel() { Channel c = channel; if (c == null || !c.isConnected()) return null; // 获取一个 NettyChannel 类型对象 return NettyChannel.getOrAddChannel(c, getUrl(), this); } } final class NettyChannel extends AbstractChannel { private static final ConcurrentMap<org.jboss.netty.channel.Channel, NettyChannel> channelMap = new ConcurrentHashMap<org.jboss.netty.channel.Channel, NettyChannel>(); private final org.jboss.netty.channel.Channel channel; /** 私有构造方法 */ private NettyChannel(org.jboss.netty.channel.Channel channel, URL url, ChannelHandler handler) { super(url, handler); if (channel == null) { throw new IllegalArgumentException("netty channel == null;"); } this.channel = channel; } static NettyChannel getOrAddChannel(org.jboss.netty.channel.Channel ch, URL url, ChannelHandler handler) { if (ch == null) { return null; } // 尝试从集合中获取 NettyChannel 实例 NettyChannel ret = channelMap.get(ch); if (ret == null) { // 如果 ret = null，则创建一个新的 NettyChannel 实例 NettyChannel nc = new NettyChannel(ch, url, handler); if (ch.isConnected()) { // 将 <Channel, NettyChannel> 键值对存入 channelMap 集合中 ret = channelMap.putIfAbsent(ch, nc); } if (ret == null) { ret = nc; } } return ret; } }

获取到 NettyChannel 实例后，即可进行后续的调用。下面看一下 NettyChannel 的 send 方法。

public void send(Object message, boolean sent) throws RemotingException { super.send(message, sent); boolean success = true; int timeout = 0; try { // 发送请求 ChannelFuture future = channel.write(message); // sent 的值源于 <dubbo:method sent="true/false" /> 中 sent 的配置值，有两种配置值： // 1. true: 等待消息发出，消息发送失败将抛出异常 // 2. false: 不等待消息发出，将消息放入 IO 队列，即刻返回 // 默认情况下 sent = false； if (sent) { timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT); // 等待消息发出，若在规定时间没能发出，success 会被置为 false success = future.await(timeout); } Throwable cause = future.getCause(); if (cause != null) { throw cause; } } catch (Throwable e) { throw new RemotingException(this, "Failed to send message ..."); } // 若 success 为 false，这里抛出异常 if (!success) { throw new RemotingException(this, "Failed to send message ..."); } }