Wu Yudong's Blog

原创文章，转载请注明： 转载自工学1号馆

这篇文章简单的介绍RPC的概念以及Hadoop RPC 框架的使用方法

RPC基本概念

RPC（Remote Procedure Call Protocol）——远程过程调用协议，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

RPC采用客户机/服务器模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。首先，客户机调用进程发送一个有进程参数的调用信息到服务进程，然后等待应答信息。在服务器端，进程保持睡眠状态直到调用信息的到达为止。当一个调用信息到达，服务器获得进程参数，计算结果，发送答复信息，然后等待下一个调用信息，最后，客户端调用进程接收答复信息，获得进程结果，然后调用执行继续进行。

Hadoop RPC基本框架

简单来说，Hadoop RPC = 动态代理 + 定制好的二进制流。如果不关注细节，从用户的角度来看，它的结构大致像下图：

Hadoop的RPC主要是通过Java的动态代理（Dynamic Proxy）与反射（Reflect）实现，源代码在org.apache.hadoop.ipc下，有以下几个主要类：

RPC：实现了一个简单的RPC模型

Client：RPC服务的客户端

Server：服务端的抽象类

RPC.Server：服务端的具体类

VersionedProtocol：所有的使用RPC服务的类都要实现该接口，在创建代理时，用来判断代理对象是否创建正确。

在后面的文章中将详细剖析各个类的源代码和运行机制

Hadoop RPC 的使用

Hadoop RPC 主要对外提供了2种接口：

public static VersionedProtocol getProxy/waitForProxy()：构建一个客户端代理对象（该对象实现了某个协议），用于向服务器发送RPC请求

public static Server getServer()：为某个协议（实际上是Java接口）实例构造一个服务器对象，用于处理客户端发送的请求

Hadoop RPC使用方法可分为以下几个步骤：

步骤1、定义RPC协议

Hadoop中所定义的RPC接口都需要继承VersionedProtocol接口，它描述了协议的版本信息

interface ClientProtocol extends org.apache.hadoop.ipc.VersionedProtocol {
	public static final long versionID = 1L;
	String echo(String value) throws IOException;
	int add(int v1, int v2) throws IOException;
}

步骤2、实现RPC协议

Hadoop RPC协议通常是一个java接口，用户需要实现该接口

public static class ClientProtocolImpl implements ClientProtocol {

	@Override
	public long getProtocolVersion(String protocol, long clientVersion)
				throws IOException {
		return ClientProtocol.versionID;
	}
	@Override
	public String echo(String value) throws IOException {
		return value;
	}

	@Override
	public int add(int v1, int v2) throws IOException {
		return v1 + v2;
	}
}

步骤3：构造并启动RPC Server

server = RPC.getServer(new ClientProtocolImpl(), serverHost, serverPort, numHandlers, false, conf);
server.start();

serverHost：服务器的host

serverPort：服务器的监听端口号

numHandlers：服务器端请求的线程数目

步骤4：构造RPC Client，并发送RPC请求

proxy = (ClientProtocol)RPC.getProxy(ClientProtocol.class, ClientProtocol.versionID, addr, conf);
int result = proxy.add(5, 6);
String echoResult = Proxy.echo("result");

在这里可以使用java动态代理的例子来简单的理解一下：

public interface CalculatorProtocol { //定义一个接口
	public int add(int a, int b);
	public int subtract(int a, int b);
}
public class DynamicProxyExample {
	public static void main(String[] args) {

		CalculatorProtocol server = new Server();
		InvocationHandler handler = new CalculatorHandler(server);
		CalculatorProtocol client = (CalculatorProtocol)Proxy.newProxyInstance(server.getClass().getClassLoader(), 
				server.getClass().getInterfaces(), handler);
		
		int r = client.add(3, 5);
		System.out.println("3 + 5 = " + r);
		r = client.subtract(7, 2);
		System.out.println("7 - 2 = " + r);
				

	}
}

class Server implements CalculatorProtocol {  //实现接口协议
	@Override
	public int add(int a, int b) {
		return a + b;
	}

	@Override
	public int subtract(int a, int b) {
		return a - b;
	}
}

class CalculatorHandler implements InvocationHandler {  //实现调用处理器接口
	private Object objOriginal;

	public CalculatorHandler(Object obj) {
		this.objOriginal = obj;
	}

	@Override
	public Object invoke(Object proxy, Method method, Object[] args)
			throws Throwable {
    //可添加一些预处理
		Object result = method.invoke(this.objOriginal, args);
    //可添加一些后续处理
		return result;
	}
}