下面我们来分析一下Dubbo的Exchange层。之前我们说过，在Protocol层进行rpc调用后，要到exchange层去封装请求响应模型，将同步转为异步，信息交换层依赖Exporter，最终将通过网络传输层接收调用请求RequestFuture和ResponseFuture。那么下面我们就来分析一下exchange层的实现。
先来看Exchanger接口
public interface Exchanger {
/**
* bind.
*
* 绑定一个服务器
*
* @param url server url
* @param handler 数据交换处理器
* @return message server 服务器
*/
@Adaptive({Constants.EXCHANGER_KEY})
ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException;
/**
* connect.
*
* 连接一个服务器，即创建一个客户端
*
* @param url server url 服务器地址
* @param handler 数据交换处理器
* @return message channel 客户端
*/
@Adaptive({Constants.EXCHANGER_KEY})
ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException;
}
再看一下HeaderExchanger这个类
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
}
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}
再来看一下ExchangeChannel接口
/**
* send request.
*
* 发送请求
*
* @param request 请求
* @return response future
* @throws RemotingException 远程调用，发生异常
*/
ResponseFuture request(Object request) throws RemotingException;
/**
* send request.
*
* 发送请求
*
* @param request 请求
* @param timeout 超时时长
* @return response future
* @throws RemotingException 远程调用，发生异常
*/
ResponseFuture request(Object request, int timeout) throws RemotingException;
/**
* get message handler.
*
* 获得信息交换处理器
*
* @return message handler
*/
ExchangeHandler getExchangeHandler();
/**
* graceful close.
*
* 优雅关闭
*
* @param timeout 超时时长
*/
void close(int timeout);
再来看一下HeaderExchangeChannel这个类
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("2.0.0");
req.setTwoWay(true); // 需要响应
req.setData(request);
// 创建 DefaultFuture 对象
DefaultFuture future = new DefaultFuture(channel, req, timeout);
try {
// 发送请求
channel.send(req);
} catch (RemotingException e) { // 发生异常，取消 DefaultFuture
future.cancel();
throw e;
}
// 返回 DefaultFuture 对象
return future;
}
在这个方法中，创建了一个DefaultFuture对象，将同步转为异步，channel调用send发送请求之后，不需要等待结果，直接将DefaultFuture返回给上层，上层可以通过调用DefaultFuture的get方法来获取响应，get方法会阻塞等待获取服务器的响应才会返回；Client接收消息在handler里面，比如Netty在NettyHandler里面messageReceived方法介绍响应消息，NettyHandler最终会调用上面传入的DecodeHandler，DecodeHandler会先判断一下是否已经解码，如果解码就直接调用HeaderExchangeHandler，默认已经设置了编码解码器，所以会直接调用HeaderExchangeHandler里面的received方法。
public void received(Channel channel, Object message) throws RemotingException {
// 设置最后的读时间
channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
// 创建 ExchangeChannel 对象
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
try {
// 处理请求( Request )
if (message instanceof Request) {
// handle request.
Request request = (Request) message;
// 处理事件请求
if (request.isEvent()) {
handlerEvent(channel, request);
} else {
// 处理普通请求
if (request.isTwoWay()) {
Response response = handleRequest(exchangeChannel, request);
channel.send(response);
// 提交给装饰的 `handler`，继续处理
} else {
handler.received(exchangeChannel, request.getData());
}
}
// 处理响应( Response )
} else if (message instanceof Response) {
handleResponse(channel, (Response) message);
// 处理 String
} else if (message instanceof String) {
// 客户端侧，不支持 String
if (isClientSide(channel)) {
Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
logger.error(e.getMessage(), e);
// 服务端侧，目前是 telnet 命令
} else {
String echo = handler.telnet(channel, (String) message);
if (echo != null && echo.length() > 0) {
channel.send(echo);
}
}
// 提交给装饰的 `handler`，继续处理
} else {
handler.received(exchangeChannel, message);
}
} finally {
// 移除 ExchangeChannel 对象，若已断开
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}
再到handleResponse方法
static void handleResponse(Channel channel, Response response) {
if (response != null && !response.isHeartbeat()) {
DefaultFuture.received(channel, response);
}
}
然后到DefaultFuture的received方法
public static void received(Channel channel, Response response) {
try {
// 移除 FUTURES
DefaultFuture future = FUTURES.remove(response.getId());
// 接收结果
if (future != null) {
future.doReceived(response);
} else {
logger.warn("The timeout response finally returned at "
+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
+ ", response " + response
+ (channel == null ? "" : ", channel: " + channel.getLocalAddress()
+ " -> " + channel.getRemoteAddress()));
}
// 移除 CHANNELS
} finally {
CHANNELS.remove(response.getId());
}
}
下面我们来看一下服务端的处理请求的流程
先来看一下接口ExchangeServer
/**
* get channels.
*
* 获得通道数组
*
* @return channels 通道数组
*/
Collection<ExchangeChannel> getExchangeChannels();
/**
* get channel.
*
* 获得指定通道
*
* @param remoteAddress 远程地址
* @return channel 通道
*/
ExchangeChannel getExchangeChannel(InetSocketAddress remoteAddress);
再来看一个类HeaderExchangeServer
public void send(Object message) throws RemotingException {
if (closed.get()) {
throw new RemotingException(this.getLocalAddress(), null, "Failed to send message " + message + ", cause: The server " + getLocalAddress() + " is closed!");
}
server.send(message);
}
然后调用HeaderExchangeHandler的handleRequest方法
Response handleRequest(ExchangeChannel channel, Request req) {
Response res = new Response(req.getId(), req.getVersion());
// 请求无法解析，返回 BAD_REQUEST 响应
if (req.isBroken()) {
Object data = req.getData();
String msg; // 请求数据，转成 msg
if (data == null) {
msg = null;
} else if (data instanceof Throwable) {
msg = StringUtils.toString((Throwable) data);
} else {
msg = data.toString();
}
res.setErrorMessage("Fail to decode request due to: " + msg);
res.setStatus(Response.BAD_REQUEST);
return res;
}
// 使用 ExchangeHandler 处理，并返回响应
// find handler by message class.
Object msg = req.getData();
try {
// handle data.
Object result = handler.reply(channel, msg);
res.setStatus(Response.OK);
res.setResult(result);
} catch (Throwable e) {
res.setStatus(Response.SERVICE_ERROR);
res.setErrorMessage(StringUtils.toString(e));
}
return res;
}
首先创建了一个Response，并且指定responseId为requestId，方便在客户端定位到具体的DefaultFuture；然后调用handler的reply方法处理消息，返回结果，如何处理的将在后面的protocol层介绍，大致就是通过Request的信息，反射调用Server端的服务，然后返回结果，然后将结果放入Response对象中，通过channel将消息发送客户端。
Duboo的Exchange层分析就到这里了。
网友评论