Hadoop 学习之RPC的使用与实现原理

cqlinx

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>();
　　
    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