Hadoop RPC
Hadoop作为分布式存储系统,为了实现各节点间的通信和交互,所以需要实现一套这样的机制.RPC(Remote Procedure CallProtocol, 远程过程调用协议)允许本地程序像调用本地方法一样调用调用远程机器上应用程序提供的服务.所以Hadoop实现了自己的RPC协议.
Hadoop RPC协议基于IPC(Inter-Process Communcations,进程间通信)实现了一套高效的轻量级RPC,这套框架使用了JAVANIO,动态代理,Protobuf
RPC架构图：
客户端代理程序将请求程序的RPC调用序列化，并调用Client.call()方法将请求发送给远程服务器，这些实现对客户端调用程序是完全透明的。
Hadoop2.x默认使用protobuf作为序列化工具，当然Hadoop框架也支持其他的序列化框架。
Hadoop定义了RpcEngine接口抽象使用不同的序列化框架的RPC引擎，RPCEngine接口包括两个重要的方法。
getProxy():采用java动态代理机制，客户端在代理对象上的调用会由RpcInvocationHandler对象处理。RpcInvocationHandler会将请求序列化，并调用Client.call()方法将请求发送到远程服务器。当服务器返回响应信息后，RpcInvocationHandler会将响应信息反序列化并返回给调用程序。
getServer():用于产生一个RPC Server对象,服务器会启动这个Server对象监听从客户端发来的请求.成功从网络接收数据后,Server对象会调用RpcInvoker对象处理这个请求.
RpcEngine目前有两个子类:
WritableRpcEngine : Hadoop自带的Writable作为序列化工具的RPC引擎.
ProtobufRpcEngine :使用protobuf作为序列化工具的RPC引擎.
Java动态代理:
//Handler
public class MyInvocationHandler implements InvocationHandler{
//被代理的对象
private Object target = null;
//通过构造方法传递一个对象
public MyInvocationHandler(Object _obj){
this.target = _obj
}
//代理方法
public invoke(Object proxy, Method method, Object[] args) throws Throwable{
//Aop before切面
//执行被代理方法
Object obj = method.invoke(this.target,args);
// Aop afte切面
}
}
//动态代理类
public class DynamicProxy<T>{
public static <T> T newProxyInstance(ClassLoader loader,Class<?>[] interfaces,
InvocationHandler h){
return (T)Proxy.newProxyIntstance(loader,interfaces,h);
}
}
//具体业务的动态代理类
public class TargetDynamicProxy extends DynamicProxy {
public static <T> T newProxyInstance(TargetInterface target){
ClassLoader loader = target.getClass().getLoader();
Class<?>[] interfaces = target.getCalss.getInterfaces();
InvocationHandler Handler = new MyInvocationHandler(target);
return newProxyIntstance(loader,interfaces, Handler);
}
}
//要代理的接口
public interface TargetInterface{
public void doSomething(String str);
}
//接口实现类
public class MyTarget implements TargetInterface{
public void doSomething(){
System.out.println(“do something!”);
}
}
//测试动态代理
public class Test{
public static void main(String[] args){
//自定义一个目标对象
TargetInterface target = new MyTarget();
//定义目标对象的代理对象
TargetInterface proxy = TargetDynamicProxy. newProxyInstance(target);
//代理目标对象执行方法
proxy.doSomething();
}
}
Rename RPC调用流程图：
Client发送请求和接收响应流程：
Client.call()方法将RPC请求封装成一个CALL对象，CALL对象中保存了RPC调用的完成标志、返回值信息以及异常信息；随后Client.call()方法会创建一个Connection对象，Connection对象用户管理Client与Server的soket链接。
用ConnectionId做为key，将新建的Connection对象放入Client.Connections字段中保存（对于Connection对象，由于涉及了与Server的Socket连接，会比较消耗资源，所以Client类使用HashTable对象connections保存哪些没有过期的Connection，如果可以复用则复用这些Connection对象）;以callId为key，将构造的call对象放入Connection.calls字段中保存。
Client.call()方法调用Connection.setupIOstreams()方法建立与Server的Socket连接。setupIOstreams()方法还会启动Connection线程，Connection线程会监听Socket并读取Server发回的响应信息。
Client.call()方法调用Connection.sendRpcRequest()发送RPC请求到Server。
Client.call()方法调用Call.wait()在Call对象上等待，等待Server发回响应信息。
Connection线程收到Server发回的响应信息，根据响应消息中携带的信息找到对应的Call对象，然后设置Call对象的返回字段，并调用call.notify()唤醒调用Client.call()方法的线程读取call()对应的返回值。
Server为了提高性能，Server类采用了很多技术来提高并发能力，包括线程池、JavaNIO提供的Reactor，其中Reactor模式贯穿整个Server的设计。
Reactor模式是一种广泛应用在服务器端的设计模式，也是一种事件驱动的设计模式。Reactor模式的处理流程是：应用程序向中间人注册IO事件，当中间人监听到这个IO事件的发生后，会通知并唤醒应用程序处理这个事件。这里的中间人其实是一个不断等待和循环的线程，它接受所有请用程序的注册，并检查应用程序注册的io事件是否就绪，如果就绪了则通知应用程序进行处理。
Server类的设计是典型的多线程Reactor的网络服务器的结构。Server定义了如下几个内部类，我们可以对Reactor结构中的模块来理解。
Listener：类似于Reactor 模型中的mainReactor。Listener对象中存在一个Selector对象acceptSelector，负责监听来自客户端的Socket连接请求。当acceptSelector监听到连接请求后，Listener对象会初始化这个连接，之后采用轮询的方式从readers线程池中选择一个reader线程处理RPC请求的读操作。
Reader：用于读取RPC请求。Reader线程类中存在一个Selector对象readSelector，这个对象用户监听网络中是否有可以读取的RPC请求。当readerSelector监听到可读的RPC请求后，会唤醒Reader线程读取这个请求，并将请求封装到一个Call对象中，然后将这个Call对象放入共享队列CallQueue中。
Handler：用户处理RPC请求并发回响应。Handler对象从CallQueue中不停地取出RPC请求，然后执行RPC请求对应的本地函数，最后封装响应并将响应发回客户端。为了能够并发的处理RPC请求，Server中会存在多个Handler对象。
Responder:用户向客户端发送RPC响应。Handler中不是已经发送RPC响应了吗？为什么还要再实现Responder类？这是出于对响应很大时或网络条件不好时，Handler线程很难将完整的响应发回客户端，这就会造成Handler线程阻塞，从而影响RPC请求的处理效率。出现这种情况时，Responder内部的respondSelector上注册写响应事件，这里的respondSelector会监听网络环境具备写响应条件时，会通知Responder将剩余响应发回客户端。
服务端处理客户请求流程：
网友评论