ChannelDuplexHandler
AbstractRpcRemoting类继承于Netty的ChannelDuplexHandler，该类实现了自己定义的ChannelInboundHandler和ChannelOutboundHandler两个接口，关于Netty定义的ChannelChandler概念 可以通过链接进行了解，这两个接口定义了处理请求及发起请求调用的API。所以ChannelDuplexHandler实现了处理请求及发启请求调用的一套基本实现。
AbstractRpcRemoting
image.png
AbstractRpcRemoting则是根据自己的业务使用场景，做了进一步的RPC封装。将RPC调用的客户端及服务端的通用实现都放在了该类中。AbstractRpcRemoting类主要实现了以下封装:
- RPC调用及响应处理都使用自定义的
RpcMessage类型数据格式。
RpcMessage为fescar自定义的RPC调用传输的数据对象，主要属性有：
//序列号
private static AtomicLong NEXT_ID = new AtomicLong(0);
public static long getNextMessageId() {
return NEXT_ID.incrementAndGet();
}
private long id;
private boolean isAsync; //是同步调用还是异步调用
private boolean isRequest; //是否为调用请求
private boolean isHeartbeat; //是否为心跳
private Object body; //传输的数据
- 对ChannelDuplexHandler类的
channelRead(ChannelHandlerContext, Object)方法覆写，实现客户端对接收到的消息进行处理。 -
sendAsyncRequest封装发起RPC调用，及获取调用结果。
下面对channelRead及sendAsyncRequest方法进行分析。
channelRead
channelRead根据数据的类型，分别做处理：
- 服务端的调用请求
处理逻辑交由dispatch(long, ChannelHandlerContext, Object)方法来处理。
AbstractRpcRemoting.this.messageExecutor.execute(new Runnable() {
@Override
public void run() {
try {
dispatch(rpcMessage.getId(), ctx, rpcMessage.getBody());
} catch (Throwable th) {
LOGGER.error(FrameworkErrorCode.NetDispatch.errCode, th.getMessage(), th);
}
}
});
dispatch是一个抽象方法，具体逻辑需要子类来实现。
public abstract void dispatch(long msgId, ChannelHandlerContext ctx, Object msg);
- 服务端的请求响应
如果通过RpcMessage对象的消息ID在futures对象中查到缓存，则认为是客户端请求的响应数据。则删除请求的缓存信息，并将请求中的数据信息设置到futures对应键值的消息结果中。
MessageFuture messageFuture = futures.remove(rpcMessage.getId());
if (messageFuture != null) {
messageFuture.setResultMessage(rpcMessage.getBody());
- 既不是服务端的请求，也不是客户端的请求响应
则也交由dispatch(rpcMessage.getId(), ctx, rpcMessage.getBody());方法进行处理。
(注：本人没有想到会有什么情况会走这个分支-_-||)
sendAsyncRequest
除了可以接收消息以外，还实现了一个重要的API就是发送异步消息。该功能的实现使用了线程池及多线程编程的Future模式。并且分为两种情况进行发送。
- 合并发送
ConcurrentHashMap<String, BlockingQueue<RpcMessage>> map = basketMap;
BlockingQueue<RpcMessage> basket = map.get(address);
if (basket == null) {
map.putIfAbsent(address, new LinkedBlockingQueue<RpcMessage>());
basket = map.get(address);
}
basket.offer(rpcMessage);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("offer message: " + rpcMessage.getBody());
}
if (!isSending) {
synchronized (mergeLock) {
mergeLock.notifyAll();
}
}
因为每次发送请求都需要与配置的对方地址创建通讯连接，然后发送消息。
但如果并发率太高，就会有性能瓶颈。
fescar的处理方式为，如果消息是向往同一个地址的请求，则将信息缓存在同一个阻塞队列中。
每隔一秒就通过异步线程从每一个阻塞队列中循环取出所有消息合并为一条消息，一次性发送。
public class MergedSendRunnable implements Runnable {
@Override
public void run() {
while (true) {
synchronized (mergeLock) {
try {
mergeLock.wait(MAX_MERGE_SEND_MILLS);
} catch (InterruptedException e) {}
}
isSending = true;
for (String address : basketMap.keySet()) {
BlockingQueue<RpcMessage> basket = basketMap.get(address);
if (basket.isEmpty()) { continue; }
MergedWarpMessage mergeMessage = new MergedWarpMessage();
while (!basket.isEmpty()) {
RpcMessage msg = basket.poll();
mergeMessage.msgs.add((AbstractMessage)msg.getBody());
mergeMessage.msgIds.add(msg.getId());
}
if (mergeMessage.msgIds.size() > 1) {
printMergeMessageLog(mergeMessage);
}
Channel sendChannel = connect(address);
try {
sendRequest(sendChannel, mergeMessage);
} catch (FrameworkException e) {
if (e.getErrcode() == FrameworkErrorCode.ChannelIsNotWritable
&& address != null) {
destroyChannel(address, sendChannel);
}
LOGGER.error("", "client merge call failed", e);
}
}
isSending = false;
}
}
这块代码感觉个人认为应该将isSending属性设置为volatile,提高跨线程的可见性。
- 直接发送
上面是有地址信息的情况，如果没有信息则直接发送，通过回调设置调用返回结果。
} else {
ChannelFuture future;
channelWriteableCheck(channel, msg);
future = channel.writeAndFlush(rpcMessage);
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
if (!future.isSuccess()) {
MessageFuture messageFuture = futures.remove(rpcMessage.getId());
if (messageFuture != null) {
messageFuture.setResultMessage(future.cause());
}
destroyChannel(future.channel());
}
}
});
}
最后判断参数timeout是否有设置，如果有则通过MessageFuture对象的get方法获取返回值，否则返回空。
if (timeout > 0) {
try {
return messageFuture.get(timeout, TimeUnit.MILLISECONDS);
} catch (Exception exx) {
LOGGER.error("wait response error:" + exx.getMessage() + ",ip:" + address + ",request:" + msg);
if (exx instanceof TimeoutException) {
throw (TimeoutException)exx;
} else {
throw new RuntimeException(exx);
}
}
} else {
return null;
}
这里要查看一下MessageFuture对象，她有一个CountDownLatch类型计数器latch对象。
private final CountDownLatch latch = new CountDownLatch(1);
当调用异步发送消息的以后，调用get(long timeout, TimeUnit unit)方法时，当前线程会处理等待状态。
boolean success = latch.await(timeout, unit);
如果异步调用完成，调用MessageFuture对象的setResultMessage(Object obj)方法时，会对计数器进行减数。减数完成以后，latch.await()将被唤醒，返回结果值。
public void setResultMessage(Object obj) {
this.resultMessage = (obj == null ? NULL : obj);
latch.countDown();
}
