Hadoop学习之RPC的使用与实现原理
hadoop作为分布式的系统, 集群机器之间的通信是最基本,最常见的需求。
这种需求本质上是IPC, 即进程间通信。 按照传统的UINX编程模型,进程间通信无非是如下的几种方式:
管道, FIFO, 消息队列, 信号量, 共享存储, 套接字。只有套接字是可以跨机器的网络通信, 能满足hadoop的需求。
通常情况下, 网络通信的程序使用显式网络编程(即直接使用java.net包)。比如Web浏览器, Web服务器等。
但也有另一部分程序使用隐式网络编程, 比如利用hadoop RPC这种封装了底层通信细节的工具包。
目前成都创新互联公司已为上1000+的企业提供了网站建设、域名、虚拟主机、网站运营、企业网站设计、柞水网站维护等服务,公司将坚持客户导向、应用为本的策略,正道将秉承"和谐、参与、激情"的文化,与客户和合作伙伴齐心协力一起成长,共同发展。
这样做使得底层的网络通信对于程序员透明。一则减轻了程序员的负担, 二则抽象了功能模块, 使得模块之间职责更清晰, 便于维护。
首先展示一个hadoop RPC功能demo, 了解hadoop RPC的用法。
step 1: 在pom.xml文件中添加依赖项
org.apache.hadoop hadoop-common 2.6.0
step 2: 在src/main/resources中添加log4j.properties
log4j.rootLogger=DEBUG,console log4j.additivity.org.apache=true # (console) log4j.appender.console=org.apache.log4j.ConsoleAppender log4j.appender.console.Threshold=INFO log4j.appender.console.ImmediateFlush=true log4j.appender.console.Target=System.err log4j.appender.console.layout=org.apache.log4j.PatternLayout log4j.appender.console.layout.ConversionPattern=[%-5p] %d(%r) --> [%t] %l: %m %x %n
step 3: 定义RPC协议
package hadooprpc.demo; import java.io.IOException; interface ClientProtocol extends org.apache.hadoop.ipc.VersionedProtocol { // 版本号,默认情况下,不同版本号的 RPC Client 和 Server 之间不能相互通信 public static final long versionID = 1L; String echo(String value) throws IOException; int add(int v1, int v2) throws IOException; }
step 4: 实现RPC协议
package hadooprpc.demo; import java.io.IOException; import org.apache.hadoop.ipc.ProtocolSignature; public class ClientProtocolImpl implements ClientProtocol { // 重载的方法,用于获取自定义的协议版本号, public long getProtocolVersion(String protocol, long clientVersion) { return ClientProtocol.versionID; } // 重载的方法,用于获取协议签名 public ProtocolSignature getProtocolSignature(String protocol, long clientVersion, int hashcode) { return new ProtocolSignature(ClientProtocol.versionID, null); } public String echo(String value) throws IOException { return value; } public int add(int v1, int v2) throws IOException { return v1 + v2; } }
step 5; 构造并启动 RPC Server
package hadooprpc.demo; import java.io.IOException; import org.apache.hadoop.HadoopIllegalArgumentException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.ipc.RPC; import org.apache.hadoop.ipc.Server; public class RPCServer { public static void main(String[] args) throws HadoopIllegalArgumentException, IOException { Configuration conf = new Configuration(); Server server = new RPC.Builder(conf).setProtocol(ClientProtocol.class) .setInstance(new ClientProtocolImpl()).setBindAddress("localhost").setPort(8097) .setNumHandlers(5).build(); server.start(); } }
step 6: 构造 RPC Client 并发送 RPC 请求
package hadooprpc.demo; import java.io.IOException; import java.net.InetSocketAddress; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.ipc.RPC; public class RPCClient { public static void main(String[] args) throws IOException { ClientProtocol client = (ClientProtocol)RPC.getProxy(ClientProtocol.class, ClientProtocol.versionID, new InetSocketAddress(8097), new Configuration()); int result = client.add(5, 6); System.out.println(result); String echoResult = client.echo("result"); System.out.println(echoResult); RPC.stopProxy(client); -- 关闭连接 } }
step 7: 启动RPC Server, 然后执行 RPC Client.
通过上面的例子可以发现, 通过这种编程方式,不用考虑网络层的细节,只需要编写接口和接口实现即可。
问渠那得清如许? 通过hadoop RPC的源码, 或许可以管中窥豹。
hadoop RPC的实现原理很简单。
Client:
1. 通过动态代理,获取到调用接口的方法,参数类型。
2. 将调用信息编码,发送到服务器
3. 获取服务器的返回值, 并解码。
4. 返回调用方法的返回值。
Server:
1. 启动服务器, 并监听客户端。
2. 获取客户端发送过来的调用方法, 参数。
3. 执行实现类中相关的方法。
4. 将返回值发送到客户端。
理解原理后, 自己动手实现了一个非常粗糙, 只具备演示功能的RPC框架。
package srpc; import java.io.*; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.lang.reflect.Proxy; import java.net.ServerSocket; import java.net.Socket; import java.net.SocketException; import java.text.MessageFormat; import java.util.HashMap; import java.util.Map; /** * Created by shgy on 17-5-7. * 创建一个简单的PRC框架, 参数和返回值只支持基础类型 */ // 接口 interface MyProtocol{ String echo(String msg); int add(int a,int b); } // 实现 class MyProtocolImp implements MyProtocol{ @Override public String echo(String msg) { return msg; } @Override public int add(int a,int b){ return a+b; } } public class ImitateRPC { private static final Map> PRIMITIVE_NAMES = new HashMap >(); static { PRIMITIVE_NAMES.put("boolean", Boolean.TYPE); PRIMITIVE_NAMES.put("byte", Byte.TYPE); PRIMITIVE_NAMES.put("char", Character.TYPE); PRIMITIVE_NAMES.put("short", Short.TYPE); PRIMITIVE_NAMES.put("int", Integer.TYPE); PRIMITIVE_NAMES.put("long", Long.TYPE); PRIMITIVE_NAMES.put("float", Float.TYPE); PRIMITIVE_NAMES.put("double", Double.TYPE); PRIMITIVE_NAMES.put("void", Void.TYPE); } public static class Server{ private Class> protocolClass; private Object protocolImpl; private ServerSocket server; public Server(Class> protocolClass, Object protocolImpl) throws IOException { if(protocolImpl == null){ throw new IllegalArgumentException("protocolImpl is not set"); } if (protocolClass == null) { throw new IllegalArgumentException("protocolClass is not set"); } else { if (!protocolClass.isAssignableFrom(protocolImpl.getClass())) { throw new IOException("protocolClass "+ protocolClass + " is not implemented by protocolImpl which is of class " + protocolImpl.getClass()); } } this.protocolClass = protocolClass; this.protocolImpl = protocolImpl; } public void start(){ System.out.println("start server"); try{ this.server = new ServerSocket(8189); listen(); }catch(Exception e){ e.printStackTrace(); } } public void close(){ System.out.println("close server"); try { this.server.close(); } catch (IOException e) {} } private void listen(){ new Thread(){ @Override public void run(){ while (!server.isClosed()){ Socket incoming = null; try { incoming = server.accept(); DataInputStream inStream = new DataInputStream(incoming.getInputStream()); // 从客户端读取出 调用方法的信息 int dataLen = inStream.readInt(); byte[] data = new byte[dataLen]; inStream.read(data,0, dataLen); DataInputStream contentStream = new DataInputStream(new ByteArrayInputStream(data)); String methodName = contentStream.readUTF(); int paramCount = contentStream.readInt(); Class>[] paramTypes = new Class>[paramCount]; Object[] args = new Object[paramCount]; for(int i=0;i declaredClass = PRIMITIVE_NAMES.get(className); if(declaredClass == null){ declaredClass = String.class; } paramTypes[i] = declaredClass; if(declaredClass == String.class){ args[i] = contentStream.readUTF(); }else if (declaredClass.isPrimitive()) { // primitive types if (declaredClass == Boolean.TYPE) { // boolean args[i] = Boolean.valueOf(contentStream.readBoolean()); } else if (declaredClass == Character.TYPE) { // char args[i] = Character.valueOf(contentStream.readChar()); } else if (declaredClass == Byte.TYPE) { // byte args[i] = Byte.valueOf(contentStream.readByte()); } else if (declaredClass == Short.TYPE) { // short args[i] = Short.valueOf(contentStream.readShort()); } else if (declaredClass == Integer.TYPE) { // int args[i] = Integer.valueOf(contentStream.readInt()); } else if (declaredClass == Long.TYPE) { // long args[i] = Long.valueOf(contentStream.readLong()); } else if (declaredClass == Float.TYPE) { // float args[i] = Float.valueOf(contentStream.readFloat()); } else if (declaredClass == Double.TYPE) { // double args[i] = Double.valueOf(contentStream.readDouble()); } else if (declaredClass == Void.TYPE) { // void args[i] = null; } else { throw new IllegalArgumentException("Not a primitive: "+declaredClass); } } } Method method = protocolClass.getMethod(methodName, paramTypes); Object obj = method.invoke(protocolImpl, args); Class retType = method.getReturnType(); ByteArrayOutputStream buffer = new ByteArrayOutputStream(); DataOutput out = new DataOutputStream(buffer); out.writeUTF(retType.getCanonicalName()); if(retType == String.class){ out.writeUTF((String)obj); }else if (retType.isPrimitive()) { // primitive types if (retType == Boolean.TYPE) { // boolean out.writeBoolean((Boolean)obj); } else if (retType == Character.TYPE) { // char out.writeChar((Character) obj); } else if (retType == Byte.TYPE) { // byte out.writeBoolean((Boolean)obj); } else if (retType == Short.TYPE) { // short out.writeShort((Short) obj); } else if (retType == Integer.TYPE) { // int out.writeInt((Integer) obj); } else if (retType == Long.TYPE) { // long out.writeLong((Long) obj); } else if (retType == Float.TYPE) { // float out.writeFloat((Float) obj); } else if (retType == Double.TYPE) { // double out.writeDouble((Double) obj); } else if (retType == Void.TYPE) { // void } else { throw new IllegalArgumentException("Not a primitive: "+retType); } } byte[] array = buffer.toByteArray(); //将返回结果写回到客户端 DataOutputStream outStream = new DataOutputStream(incoming.getOutputStream()); outStream.writeInt(array.length); outStream.write(array); outStream.flush(); }catch (SocketException e){ }catch (Exception e){ e.printStackTrace(); }finally { try { if(incoming!=null){ incoming.close(); } }catch (Exception e){} } } } }.start(); } } public static T getProxy(Class protocol){ T proxy = (T) Proxy.newProxyInstance(protocol.getClassLoader(), new Class[]{protocol}, new InvocationHandler() { @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { Class>[] paramTypes = method.getParameterTypes(); ByteArrayOutputStream buffer = new ByteArrayOutputStream(); DataOutput out = new DataOutputStream(buffer); out.writeUTF(method.getName()); // 写入方法名 out.writeInt(paramTypes.length); // 写入参数个数 // 写入参数 for (int i = 0; i < paramTypes.length; i++) { Class> declaredClass = paramTypes[i]; Object instance = args[i]; String paramTypeName = declaredClass.getCanonicalName(); out.writeUTF(paramTypeName); if (declaredClass == String.class) { out.writeUTF((String)instance); }else if (declaredClass.isPrimitive()) { // primitive type if (declaredClass == Boolean.TYPE) { // boolean out.writeBoolean(((Boolean) instance).booleanValue()); } else if (declaredClass == Character.TYPE) { // char out.writeChar(((Character) instance).charValue()); } else if (declaredClass == Byte.TYPE) { // byte out.writeByte(((Byte) instance).byteValue()); } else if (declaredClass == Short.TYPE) { // short out.writeShort(((Short) instance).shortValue()); } else if (declaredClass == Integer.TYPE) { // int out.writeInt(((Integer) instance).intValue()); } else if (declaredClass == Long.TYPE) { // long out.writeLong(((Long) instance).longValue()); } else if (declaredClass == Float.TYPE) { // float out.writeFloat(((Float) instance).floatValue()); } else if (declaredClass == Double.TYPE) { // double out.writeDouble(((Double) instance).doubleValue()); } else if (declaredClass == Void.TYPE) { // void } else { throw new IllegalArgumentException("Not a primitive: " + declaredClass); } }else{ throw new IOException("Can't write: "+instance+" as "+declaredClass); } } // 发送到服务器端 byte[] array = buffer.toByteArray(); Socket client = new Socket("127.0.0.1", 8189); try{ DataOutputStream outStream = new DataOutputStream(client.getOutputStream()); outStream.writeInt(array.length); outStream.write(array); outStream.flush(); // 读取服务器的应答 DataInputStream inStream = new DataInputStream(client.getInputStream()); int bufSize = inStream.readInt(); byte[] retArray = new byte[bufSize]; inStream.read(retArray,0, bufSize); DataInputStream contentStream = new DataInputStream(new ByteArrayInputStream(retArray)); String retType = contentStream.readUTF(); Class retTypeClass = PRIMITIVE_NAMES.get(retType); if(retTypeClass == null){ retTypeClass = String.class; } if(retTypeClass == String.class){ return contentStream.readUTF(); }else if (retTypeClass.isPrimitive()) { // primitive types if (retTypeClass == Boolean.TYPE) { // boolean return Boolean.valueOf(contentStream.readBoolean()); } else if (retTypeClass == Character.TYPE) { // char return Character.valueOf(contentStream.readChar()); } else if (retTypeClass == Byte.TYPE) { // byte return Byte.valueOf(contentStream.readByte()); } else if (retTypeClass == Short.TYPE) { // short return Short.valueOf(contentStream.readShort()); } else if (retTypeClass == Integer.TYPE) { // int return Integer.valueOf(contentStream.readInt()); } else if (retTypeClass == Long.TYPE) { // long return Long.valueOf(contentStream.readLong()); } else if (retTypeClass == Float.TYPE) { // float return Float.valueOf(contentStream.readFloat()); } else if (retTypeClass == Double.TYPE) { // double return Double.valueOf(contentStream.readDouble()); } else if (retTypeClass == Void.TYPE) { // void return null; } else { throw new IllegalArgumentException("Not a primitive: "+retTypeClass); } } }finally { try { client.close(); }catch (Exception e){} } return null; } }); return proxy; } public static void main(String[] args) throws IOException { // start server Server server = new Server(MyProtocol.class, new MyProtocolImp()); server.start(); // call rpc MyProtocol client = ImitateRPC.getProxy(MyProtocol.class); System.out.println(client.echo("hello world")); System.out.println(client.add(1, 4)); server.close(); } }
上面的代码只有300多行, 而且在编码/解码部分有重叠的代码, 可以优化缩减, 功能也不完善。 但是仿照Hadoop RPC, 演示了RPC的原理。 对照着Hadoop RPC的源码, 自己编码实现, 对于源码中各个类的功能, 会有更清晰的理解。
当前文章:Hadoop学习之RPC的使用与实现原理
文章源于:http://scyanting.com/article/jdgchd.html