Hadoop 學習之RPC的使用與實現原理
hadoop做爲分佈式的系統， 集羣機器之間的通訊是最基本，最多見的需求。
這種需求本質上是IPC, 即進程間通訊。 按照傳統的UINX編程模型，進程間通訊無非是以下的幾種方式:
管道， FIFO， 消息隊列， 信號量， 共享存儲， 套接字。只有套接字是能夠跨機器的網絡通訊， 能知足hadoop的需求。
一般狀況下， 網絡通訊的程序使用顯式網絡編程（即直接使用java.net包）。好比Web瀏覽器， Web服務器等。
但也有另外一部分程序使用隱式網絡編程， 好比利用hadoop RPC這種封裝了底層通訊細節的工具包。java
這樣作使得底層的網絡通訊對於程序員透明。一則減輕了程序員的負擔， 二則抽象了功能模塊， 使得模塊之間職責更清晰， 便於維護。
首先展現一個hadoop RPC功能demo， 瞭解hadoop RPC的用法。
step 1: 在pom.xml文件中添加依賴項程序員
<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.6.0</version> </dependency>
step 2: 在src/main/resources中添加log4j.properties
apache
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<String, Class<?>> PRIMITIVE_NAMES = new HashMap<String, Class<?>>();
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<paramCount;i++){
String className = contentStream.readUTF();
Class<?> 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> T getProxy(Class<T> 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的源碼， 本身編碼實現， 對於源碼中各個類的功能， 會有更清晰的理解。app
相關文章
- 1. RPC原理與實現
- 2. RPC實現原理
- 3. RPC的實現原理
- 4. rabbbitmq實現rpc的原理
- 5. Hadoop的RPC工作原理
- 6. RPC原理的學習
- 7. RPC原理及實現 RPC原理及實現
- 8. Hadoop學習之路(三)：Hadoop排序之二次排序的原理及實現
- 9. 服務框架學習(二)：遠程方法調用(RPC)的原理與實現
- 10. RPC框架原理與實現
- 更多相關文章...
- • 使用Redis和Lua的原子性實現搶紅包功能 - 紅包項目實戰
- • MyBatis的工作原理 - MyBatis教程
- • Java Agent入門實戰（三）-JVM Attach原理與使用
- • 適用於PHP初學者的學習線路和建議
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 跳槽面試的幾個實用小技巧，不妨看看！
- 2. Mac實用技巧 |如何使用Mac系統中自帶的預覽工具將圖片變成黑白色?
- 3. Mac實用技巧 |如何使用Mac系統中自帶的預覽工具將圖片變成黑白色?
- 4. 如何使用Mac系統中自帶的預覽工具將圖片變成黑白色?
- 5. Mac OS非兼容Windows軟件運行解決方案——「以VMware & Microsoft Access爲例「
- 6. 封裝 pyinstaller -F -i b.ico excel.py
- 7. 數據庫作業三ER圖待完善
- 8. nvm安裝使用低版本node.js(非命令安裝)
- 9. 如何快速轉換圖片格式
- 10. 將表格內容分條轉換爲若干文檔
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. RPC原理與實現
- 2. RPC實現原理
- 3. RPC的實現原理
- 4. rabbbitmq實現rpc的原理
- 5. Hadoop的RPC工作原理
- 6. RPC原理的學習
- 7. RPC原理及實現 RPC原理及實現
- 8. Hadoop學習之路(三)：Hadoop排序之二次排序的原理及實現
- 9. 服務框架學習(二)：遠程方法調用(RPC)的原理與實現
- 10. RPC框架原理與實現