一起学RPC(三)
在上一篇文章中讲到jupiter的传输模块transport中的编解码器的实现。对server来言，编解码器扮演着一头一尾的门卫角色，保证进来的人是干净的，也得保证出去的人也是干净的。当然这么比喻很不恰当，但是也想不到别的比喻了。
编解码器固然重要，但是没有核心的业务处理器也没多大意义。本文的重点就是核心处理器：AcceptorHandler.
@ChannelHandler.Sharable
public class AcceptorHandler extends ChannelInboundHandlerAdapter {
private ProviderProcessor processor;
// ...
public ProviderProcessor processor() {
return processor;
}
public void processor(ProviderProcessor processor) {
this.processor = processor;
}
}
要实现一个handler很容易，直接继承ChannelInboundHandlerAdapter就行了。当然这是针对server来说的。根据类名来看，正常情况下首先会联想到肯定会有个与之对应的。针对server来说，要处理的就是入站数据，使用inbound就行了。如果有更加复杂的逻辑处理，可以去看看官方文档中其他的派生类。同时，这个实例也是能够被共享的，道理也很简单：没有状态。也许你会问：这里分明是有成员变量的呀。但是，这个成员变量是不会被改变的。如果发生了变化，这个程序设计上就有问题。理论上是绝对不允许改变的。
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Channel ch = ctx.channel();
if (msg instanceof JRequestPayload) {
JChannel jChannel = NettyChannel.attachChannel(ch);
try {
processor.handleRequest(jChannel, (JRequestPayload) msg);
} catch (Throwable t) {
processor.handleException(jChannel, (JRequestPayload) msg, Status.SERVER_ERROR, t);
}
} else {
logger.warn("Unexpected message type received: {}, channel: {}.", msg.getClass(), ch);
ReferenceCountUtil.release(msg);
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
int count = channelCounter.incrementAndGet();
logger.info("Connects with {} as the {}th channel.", ctx.channel(), count);
super.channelActive(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
int count = channelCounter.getAndDecrement();
logger.warn("Disconnects with {} as the {}th channel.", ctx.channel(), count);
super.channelInactive(ctx);
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
Channel ch = ctx.channel();
ChannelConfig config = ch.config();
// 高水位线: ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK
// 低水位线: ChannelOption.WRITE_BUFFER_LOW_WATER_MARK
if (!ch.isWritable()) {
// 当前channel的缓冲区(OutboundBuffer)大小超过了WRITE_BUFFER_HIGH_WATER_MARK
if (logger.isWarnEnabled()) {
logger.warn("{} is not writable, high water mask: {}, the number of flushed entries that are not written yet: {}.",
ch, config.getWriteBufferHighWaterMark(), ch.unsafe().outboundBuffer().size());
}
config.setAutoRead(false);
} else {
// 曾经高于高水位线的OutboundBuffer现在已经低于WRITE_BUFFER_LOW_WATER_MARK了
if (logger.isWarnEnabled()) {
logger.warn("{} is writable(rehabilitate), low water mask: {}, the number of flushed entries that are not written yet: {}.",
ch, config.getWriteBufferLowWaterMark(), ch.unsafe().outboundBuffer().size());
}
config.setAutoRead(true);
}
}
可以看到，AcceptorHandler重写了几个方法。注意，重写channelRead()方法的时候记得调用ReferenceCountUtil.release(msg)。
其中最核心的逻辑在channelRead()中处理。无非就是将解码器中反序列化后的对象进行处理罢了。当然这里接受的仅仅是JRequestPayload类型。然后将Netty的原生Channel转化为自定义的JChannel类型。这样做的目的是为了将api统一，方便接入其他网络库实现。也就说如果要换别的网络框架如mina，不需要去改动我业务的代码，只需针对别的网络库的api进行编码即可。然后使用ProviderProcessor来处理具体的业务逻辑。这个接口中提供了两个操作：handleRequest和 handleException.因此具体的业务逻辑处理全部都传递给ProviderProcessor实现了。
而channelWritabilityChanged方法在可写状态发生变化的时候会被调用。可以通过Channel#isWritable()方法来获取状态。而这里对其重写是为了判断OutboundBuffer的大小有没有超过高水位线，这里的水位线是在ChannelConfig中设置的，server初始化的时候。超过高水位线就不允许自动去读数据了。这里有一点疑惑，不清楚为什么需要调用config.setAutoRead(false)。一个比较模糊的概念是Netty的写动作并不是直接向socket中写，而是写到Netty中的缓冲区中，这个缓冲区叫做ChannelOutboundBuffer，而这个buffer的实现是使用的无界链表，如果对方的接受太慢，就会导致这个链表无限大，最坏情况会导致OOM。因此提供一种机制：设置水位线。如果超过水位线就让用户来自己决定怎么处理，具体做法就是调用channelWritabilityChanged方法。这里的这个方法将自动读关闭了，这里面大有玄机。其实是利用了TCP的滑动窗口来控制的。
比如咱俩喝酒, 你喝完一杯我就立刻给你满上, 最终你喝不动了 ,
不再举起杯子…. 你的杯子一直是满的, 我也没法继续给你倒酒
这个栗子很形象地解释了滑动窗口。结合这个动画更直观。
这里有一篇文章值得参考。
水位线
接下来要讨论的是这个processor到底是怎么处理消息的。
private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultProviderProcessor.class);
private final CloseableExecutor executor;
public DefaultProviderProcessor() {
this(ProviderExecutors.executor());
}
public DefaultProviderProcessor(CloseableExecutor executor) {
this.executor = executor;
}
@Override
public void handleRequest(JChannel channel, JRequestPayload requestPayload) throws Exception {
MessageTask task = new MessageTask(this, channel, new JRequest(requestPayload));
if (executor == null) {
task.run();
} else {
executor.execute(task);
}
}
其实不难想到，handleRequest方法中将接受到的数据做了一层封装，然后丢给线程池去处理。在Netty中，业务处理逻辑绝对不能放在IO线程中执行。IO线程只负责读取/发送数据，不能进行业务处理。这是因为如果业务逻辑中有耗时的操作就会将IO线程阻塞住，这样正常的请求也就被阻塞了，影响应用的性能。而这里的线程池也被自定义了。
CloseableExecutor是一个接口类型，正真的实现类是通过SPI机制由工厂创建出来的。关于SPI机制这里不会展开，将单独去整理一篇文章来说明。这种机制在很多框架中都有体现。
而包装类MessageTask的实现就很关键了。既然这个对象能放到线程池中，那么一定是一个Runnable或者Callable的实现。
@Override
public void run() {
// stack copy
final DefaultProviderProcessor _processor = processor;
final JRequest _request = request;
// 全局流量控制
ControlResult ctrl = _processor.flowControl(_request);
if (!ctrl.isAllowed()) {
rejected(Status.APP_FLOW_CONTROL, new JupiterFlowControlException(String.valueOf(ctrl)));
return;
}
MessageWrapper msg;
try {
JRequestPayload _requestPayload = _request.payload();
byte s_code = _requestPayload.serializerCode();
Serializer serializer = SerializerFactory.getSerializer(s_code);
// 在业务线程中反序列化, 减轻IO线程负担
if (CodecConfig.isCodecLowCopy()) {
InputBuf inputBuf = _requestPayload.inputBuf();
msg = serializer.readObject(inputBuf, MessageWrapper.class);
} else {
byte[] bytes = _requestPayload.bytes();
msg = serializer.readObject(bytes, MessageWrapper.class);
}
_requestPayload.clear();
_request.message(msg);
} catch (Throwable t) {
rejected(Status.BAD_REQUEST, new JupiterBadRequestException("reading request failed", t));
return;
}
// 查找服务
final ServiceWrapper service = _processor.lookupService(msg.getMetadata());
if (service == null) {
rejected(Status.SERVICE_NOT_FOUND, new JupiterServiceNotFoundException(String.valueOf(msg)));
return;
}
// provider私有流量控制
FlowController<JRequest> childController = service.getFlowController();
if (childController != null) {
ctrl = childController.flowControl(_request);
if (!ctrl.isAllowed()) {
rejected(Status.PROVIDER_FLOW_CONTROL, new JupiterFlowControlException(String.valueOf(ctrl)));
return;
}
}
// processing
Executor childExecutor = service.getExecutor();
if (childExecutor == null) {
process(service);
} else {
// provider私有线程池执行
childExecutor.execute(new Runnable() {
@Override
public void run() {
process(service);
}
});
}
}
这段代码十分简洁。首先将全局变量赋值为局部变量，我依稀记得在一个老外的代码中看到过，目的大概是为了节省性能。接下来就是全局流量控制，所谓的流量控制简单理解为防止请求太猛导致服务垮掉。有全局的就一定有局部的。而局部的控制是针对rpc中对外暴露某个服务。其粒度更小一点。然后就是反序列化了，这个过程在编解码器中也能完成，但是作者并没有这么做。目的也很简单，毕竟序列化是比较耗性能的，再说了，编解码器实际上也是在IO线程中处理的。这么做也是为了减轻IO线程负担。紧接着就是将序列化后的对象中的ServiceMetadata取出来 ，通过这个对象去本地容器中找ServiceWrapper。本地容器就是rpc的provider在发布一个服务到注册中心的时候本地也保存一个这个服务的相关信息。所谓的服务简单理解为就是一个service bean。而这个容器简单理解为就是一个map映射，ServiceMetadata为key，ServiceWrapper为value。找到这个ServiceWrapper后就很好办了，接下来就是调用这个service了。但是这个ServiceWrapper不仅仅是一个service，里面有很多额外的功能，比方可以有一个私有的线程池。如果有，那么在具体调用这个service的时候会使用这个私有的线程池。这种场景我反正没见过，但不排除有这种情况。如果没有那就很简单了，直接处理。这个处理过程也很容易想到，无非就是将要调用的service执行一遍，将结果序列化，再写出去。然而里面的代码实现可没有那么简单。
具体的处理逻辑全部都在process方法中。当然，在看具体实现之前很有必要对其中几个核类如ServiceWrapper和MessageWrapper进行解读。
MessageWrapper可以简单理解为调用者发送的数据。包含要调用的对象，对象的方法，方法的参数。当然实际上比这些内容要复杂很多，比如链路追踪id等。有个最重要的参数ServiceMetadata：
public class ServiceMetadata extends Directory implements Serializable {
private static final long serialVersionUID = -8908295634641380163L;
private String group; // 服务组别
private String serviceProviderName; // 服务名称
private String version; // 服务版本号
// ...
}
Directory是一个抽象类。这个命名也很容易理解，顾名思义Directory是目录的意思。对于某个服务来说，单纯的知道服务名就足矣完成调用。但是在复杂的场景下，有成百上千个服务，要做到准确调用就得对其进行分类了。而且有时候还有同一个服务版本也不一样的情形，因此版本号也得作为这个目录中的某个层级。为什么成为metadata呢？这个属性在ServiceWrapper对象中也有。可以猜到了，一定是一一对应起来的。通俗解释来说ServiceWrapper是属于服务端的。也就是说provider发布本地服务到注册中心的同时，仅仅是将元数据发出去了，注册中心有了还不能完事，自己本地得确实存在呀，不然consumer从注册中心中拿到服务元数据了去provider里找不到这个服务，这非得骂娘不可。而ServiceWrapper正是将需要暴露出去的服务在本地存起来。仅此而已。其中最重要的属性就是服务对象serviceProvider了，对于rpc来说，consumer知道的仅仅是接口，正真去干活的还是实现类。而其余的花里胡哨的东西还是有点用的，得分场景。
public class ServiceWrapper implements Serializable {
private static final long serialVersionUID = 6690575889849847348L;
// 服务元数据
private final ServiceMetadata metadata;
// 服务对象
private final Object serviceProvider;
// 服务拦截器
private final ProviderInterceptor[] interceptors;
// key: method name
// value: pair.first: 方法参数类型(用于根据JLS规则实现方法调用的静态分派)
// pair.second: 方法显式声明抛出的异常类型
private final Map<String, List<Pair<Class<?>[], Class<?>[]>>> extensions;
// 权重 hashCode() 与 equals() 不把weight计算在内
private int weight = JConstants.DEFAULT_WEIGHT;
// provider私有线程池
private Executor executor;
// provider私有流量控制器
private FlowController<JRequest> flowController;
// ...
}
而具体的核心处理逻辑process以后再慢慢看。
